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:
Andrés de la Peña 2021-03-01 12:33:02 +00:00
parent a311109e31 865b67b21d
commit 0541c51996
7 changed files with 563 additions and 199 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,5 +1,6 @@
3.11.11
Merged from 3.0:
* Fix ColumnFilter behaviour to prevent digest mitmatches during upgrades (CASSANDRA-16415)
* Update debian packaging for python3 (CASSANDRA-16396)
* Avoid pushing schema mutations when setting up distributed system keyspaces locally (CASSANDRA-16387)
Merged from 2.2:

2
build.xml
View File

@ -422,7 +422,7 @@
<dependency groupId="org.apache.thrift" artifactId="libthrift" version="0.9.2">
<exclusion groupId="commons-logging" artifactId="commons-logging"/>
</dependency>
<dependency groupId="junit" artifactId="junit" version="4.6" />
<dependency groupId="junit" artifactId="junit" version="4.12" />
<dependency groupId="org.mockito" artifactId="mockito-core" version="3.2.4" />
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.7" />
<dependency groupId="org.reflections" artifactId="reflections" version="0.9.12" />

84
src/java/org/apache/cassandra/db/filter/ColumnFilter.java
View File

@ -23,6 +23,9 @@ import java.util.*;
import com.google.common.collect.SortedSetMultimap;
import com.google.common.collect.TreeMultimap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.*;
@ -62,6 +65,8 @@ import org.apache.cassandra.net.MessagingService;
*/
public class ColumnFilter
{
private final static Logger logger = LoggerFactory.getLogger(ColumnFilter.class);
public static final Serializer serializer = new Serializer();
// True if _fetched_ is all the columns, in which case metadata must not be null. If false,
@ -111,7 +116,19 @@ public class ColumnFilter
*/
public static ColumnFilter selection(CFMetaData metadata, PartitionColumns queried)
{
return new ColumnFilter(true, metadata.partitionColumns(), queried, null);
// When fetchAll is enabled on pre CASSANDRA-10657 (3.4-), queried columns are not considered at all, and it
// is assumed that all columns are queried. CASSANDRA-10657 (3.4+) brings back skipping values of columns
// which are not in queried set when fetchAll is enabled. That makes exactly the same filter being
// interpreted in a different way on 3.4- and 3.4+.
//
// Moreover, there is no way to convert the filter with fetchAll and queried != null so that it is
// interpreted the same way on 3.4- because that Cassandra version does not support such filtering.
//
// In order to avoid inconsitencies in data read by 3.4- and 3.4+ we need to avoid creation of incompatible
// filters when the cluster contains 3.4- nodes. We do that by forcibly setting queried to null.
//
// see CASSANDRA-10657, CASSANDRA-15833, CASSANDRA-16415
return new ColumnFilter(true, metadata.partitionColumns(), Gossiper.instance.isAnyNodeOn30() ? null : queried, null);
}
/**
@ -357,9 +374,12 @@ public class ColumnFilter
s.put(subSelection.column().name, subSelection);
}
// see CASSANDRA-15833
if (isFetchAll && Gossiper.instance.isAnyNodeOn30())
// See the comment in {@link ColumnFilter#selection(CFMetaData, PartitionColumns)}
if (isFetchAll && queried != null && Gossiper.instance.isAnyNodeOn30())
{
logger.trace("Column filter will be automatically converted to query all columns because 3.0 nodes are present in the cluster");
queried = null;
}
return new ColumnFilter(isFetchAll, isFetchAll ? metadata.partitionColumns() : null, queried, s);
}
@ -385,44 +405,29 @@ public class ColumnFilter
@Override
public String toString()
{
String prefix = "";
if (isFetchAll && queried == null)
return "*/*";
if (isFetchAll)
return "*";
prefix = "*/";
if (queried.isEmpty())
return "";
return prefix + "[]";
Iterator<ColumnDefinition> defs = queried.selectOrderIterator();
if (!defs.hasNext())
return "<none>";
StringBuilder sb = new StringBuilder();
while (defs.hasNext())
StringJoiner joiner = new StringJoiner(", ", "[", "]");
Iterator<ColumnDefinition> it = queried.selectOrderIterator();
while (it.hasNext())
{
appendColumnDef(sb, defs.next());
if (defs.hasNext())
sb.append(", ");
}
return sb.toString();
}
ColumnDefinition column = it.next();
SortedSet<ColumnSubselection> s = subSelections != null ? subSelections.get(column.name) : Collections.emptySortedSet();
private void appendColumnDef(StringBuilder sb, ColumnDefinition column)
{
if (subSelections == null)
{
sb.append(column.name);
return;
if (s.isEmpty())
joiner.add(String.valueOf(column.name));
else
s.forEach(subSel -> joiner.add(String.format("%s%s", column.name , subSel)));
}
SortedSet<ColumnSubselection> s = subSelections.get(column.name);
if (s.isEmpty())
{
sb.append(column.name);
return;
}
int i = 0;
for (ColumnSubselection subSel : s)
sb.append(i++ == 0 ? "" : ", ").append(column.name).append(subSel);
return prefix + joiner.toString();
}
public static class Serializer
@ -505,9 +510,16 @@ public class ColumnFilter
}
}
// See CASSANDRA-15833
if (version <= MessagingService.VERSION_3014 && isFetchAll)
// Since nodes with and without CASSANDRA-10657 are not distinguishable by messaging version, we need to
// check whether there are any nodes running pre CASSANDRA-10657 Cassandra and apply conversion to the
// column filter so that it is interpreted in the same way as on the nodes without CASSANDRA-10657.
//
// See the comment in {@link ColumnFilter#selection(CFMetaData, PartitionColumns)}
if (isFetchAll && queried != null && Gossiper.instance.isAnyNodeOn30())
{
logger.trace("Deserialized column filter will be automatically converted to query all columns because 3.0 nodes are present in the cluster");
queried = null;
}
return new ColumnFilter(isFetchAll, fetched, queried, subSelections);
}

6
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -1833,4 +1833,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
stop();
ExecutorUtils.shutdownAndWait(timeout, unit, executor);
}
@VisibleForTesting
public void setAnyNodeOn30(boolean anyNodeOn30)
{
this.anyNodeOn30 = anyNodeOn30;
}
}

113
test/distributed/org/apache/cassandra/distributed/test/ReadDigestConsistencyTest.java Normal file
View File

@ -0,0 +1,113 @@
/*
* 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.Arrays;
import java.util.UUID;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.ICoordinator;
public class ReadDigestConsistencyTest extends TestBaseImpl
{
public static final String TABLE_NAME = "tbl";
public static final String CREATE_TABLE = String.format("CREATE TABLE %s.%s (" +
" k int, " +
" c int, " +
" s1 int static, " +
" s2 set<int> static, " +
" v1 int, " +
" v2 set<int>, " +
" PRIMARY KEY (k, c))", KEYSPACE, TABLE_NAME);
public static final String INSERT = String.format("INSERT INTO %s.%s (k, c, s1, s2, v1, v2) VALUES (?, ?, ?, ?, ?, ?)", KEYSPACE, TABLE_NAME);
public static final String SELECT_TRACE = "SELECT activity FROM system_traces.events where session_id = ? and source = ? ALLOW FILTERING;";
@Test
public void testDigestConsistency() throws Exception
{
try (Cluster cluster = init(builder().withNodes(2).start()))
{
cluster.schemaChange(CREATE_TABLE);
insertData(cluster.coordinator(1));
testDigestConsistency(cluster.coordinator(1));
testDigestConsistency(cluster.coordinator(2));
}
}
public static void checkTraceForDigestMismatch(ICoordinator coordinator, String query, Object... boundValues)
{
UUID sessionId = UUID.randomUUID();
coordinator.executeWithTracing(sessionId, query, ConsistencyLevel.ALL, boundValues);
Object[][] results = coordinator.execute(SELECT_TRACE,
ConsistencyLevel.ALL,
sessionId,
coordinator.instance().broadcastAddress().getAddress());
for (Object[] result : results)
{
String activity = (String) result[0];
Assert.assertFalse(String.format("Found Digest Mismatch while executing query: %s with bound values %s on %s/%s",
query,
Arrays.toString(boundValues),
coordinator.instance().broadcastAddress(),
coordinator.instance().getReleaseVersionString()),
activity.toLowerCase().contains("mismatch for key"));
}
}
public static void insertData(ICoordinator coordinator)
{
coordinator.execute(String.format("INSERT INTO %s.%s (k, c, s1, s2, v1, v2) VALUES (1, 1, 2, {1, 2, 3, 4, 5}, 3, {6, 7, 8, 9, 10})", KEYSPACE, TABLE_NAME), ConsistencyLevel.ALL);
coordinator.execute(String.format("INSERT INTO %s.%s (k, c, s1, s2, v1, v2) VALUES (1, 2, 3, {2, 3, 4, 5, 6}, 4, {7, 8, 9, 10, 11})", KEYSPACE, TABLE_NAME), ConsistencyLevel.ALL);
}
public static void testDigestConsistency(ICoordinator coordinator)
{
String queryPattern = "SELECT %s FROM %s.%s WHERE %s";
String[] columnss1 = {
"k, c",
"*",
"v1",
"v2",
"v1, s1",
"v1, s2"
};
String[] columnss2 = {
"s1",
"s2"
};
for (String columns : columnss1)
{
checkTraceForDigestMismatch(coordinator, String.format(queryPattern, columns, KEYSPACE, TABLE_NAME, "k = 1"));
checkTraceForDigestMismatch(coordinator, String.format(queryPattern, columns, KEYSPACE, TABLE_NAME, "k = 1 AND c = 2"));
}
for (String columns : columnss2)
{
checkTraceForDigestMismatch(coordinator, String.format(queryPattern, columns, KEYSPACE, TABLE_NAME, "k = 1"));
}
}
}

65
test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeReadTest.java
View File

@ -18,34 +18,19 @@
package org.apache.cassandra.distributed.upgrade;
import java.util.UUID;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.UpgradeableCluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.impl.DelegatingInvokableInstance;
import org.apache.cassandra.distributed.shared.DistributedTestBase;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.shared.Versions;
import org.apache.cassandra.gms.Gossiper;
import static org.apache.cassandra.distributed.test.ReadDigestConsistencyTest.CREATE_TABLE;
import static org.apache.cassandra.distributed.test.ReadDigestConsistencyTest.insertData;
import static org.apache.cassandra.distributed.test.ReadDigestConsistencyTest.testDigestConsistency;
public class MixedModeReadTest extends UpgradeTestBase
{
public static final String TABLE_NAME = "tbl";
public static final String CREATE_TABLE = String.format(
"CREATE TABLE %s.%s (key int, c1 text, c2 text, c3 text, PRIMARY KEY (key))",
DistributedTestBase.KEYSPACE, TABLE_NAME);
public static final String INSERT = String.format(
"INSERT INTO %s.%s (key, c1, c2, c3) VALUES (?, ?, ?, ?)",
DistributedTestBase.KEYSPACE, TABLE_NAME);
public static final String SELECT_C1 = String.format("SELECT key, c1 FROM %s.%s WHERE key = ?",
DistributedTestBase.KEYSPACE, TABLE_NAME);
public static final String SELECT_TRACE = "SELECT activity FROM system_traces.events where session_id = ? and source = ? ALLOW FILTERING;";
@Test
public void mixedModeReadColumnSubsetDigestCheck() throws Throwable
{
@ -56,47 +41,25 @@ public class MixedModeReadTest extends UpgradeTestBase
.withConfig(config -> config.with(Feature.GOSSIP, Feature.NETWORK))
.setup(cluster -> {
cluster.schemaChange(CREATE_TABLE);
cluster.coordinator(1).execute(INSERT, ConsistencyLevel.ALL, 1, "foo", "bar", "baz");
cluster.coordinator(1).execute(INSERT, ConsistencyLevel.ALL, 2, "foo", "bar", "baz");
// baseline to show no digest mismatches before upgrade
checkTraceForDigestMismatch(cluster, 1);
checkTraceForDigestMismatch(cluster, 2);
insertData(cluster.coordinator(1));
testDigestConsistency(cluster.coordinator(1));
testDigestConsistency(cluster.coordinator(2));
})
.runAfterNodeUpgrade((cluster, node) -> {
if (node != 1)
return; // shouldn't happen but guard for future test changes
.runAfterClusterUpgrade(cluster -> {
// we need to let gossip settle or the test will fail
int attempts = 1;
while (!((DelegatingInvokableInstance) (cluster.get(1))).delegate().callOnInstance(() -> Gossiper.instance.isAnyNodeOn30()))
//noinspection Convert2MethodRef
while (!((IInvokableInstance) (cluster.get(1))).callOnInstance(() -> Gossiper.instance.isAnyNodeOn30()))
{
if (attempts > 30)
if (attempts++ > 30)
throw new RuntimeException("Gossiper.instance.isAnyNodeOn30() continually returns false despite expecting to be true");
Thread.sleep(1000);
}
// should not cause a disgest mismatch in mixed mode
checkTraceForDigestMismatch(cluster, 1);
checkTraceForDigestMismatch(cluster, 2);
testDigestConsistency(cluster.coordinator(1));
testDigestConsistency(cluster.coordinator(2));
})
.run();
}
private void checkTraceForDigestMismatch(UpgradeableCluster cluster, int coordinatorNode)
{
UUID sessionId = UUID.randomUUID();
cluster.coordinator(coordinatorNode).executeWithTracing(sessionId, SELECT_C1, ConsistencyLevel.ALL, 1);
Object[][] results = cluster.coordinator(coordinatorNode)
.execute(SELECT_TRACE, ConsistencyLevel.ALL,
sessionId, cluster.get(coordinatorNode).broadcastAddress().getAddress());
for (Object[] result : results)
{
String activity = (String) result[0];
Assert.assertFalse("Found Digest Mismatch", activity.toLowerCase().contains("mismatch for key"));
}
}
}

491
test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java
View File

@ -18,16 +18,28 @@
package org.apache.cassandra.db.filter;
import org.junit.Test;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.function.Consumer;
import com.google.common.base.Throwables;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import junit.framework.Assert;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.PartitionColumns;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
@ -36,143 +48,400 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import static org.junit.Assert.assertEquals;
@RunWith(Parameterized.class)
public class ColumnFilterTest
{
private static final ColumnFilter.Serializer serializer = new ColumnFilter.Serializer();
@Test
public void columnFilterSerialisationRoundTrip() throws Exception
private final CFMetaData metadata = CFMetaData.Builder.create("ks", "table")
.withPartitioner(Murmur3Partitioner.instance)
.addPartitionKey("pk", Int32Type.instance)
.addClusteringColumn("ck", Int32Type.instance)
.addStaticColumn("s1", Int32Type.instance)
.addStaticColumn("s2", SetType.getInstance(Int32Type.instance, true))
.addRegularColumn("v1", Int32Type.instance)
.addRegularColumn("v2", SetType.getInstance(Int32Type.instance, true))
.build();
private final ColumnDefinition s1 = metadata.getColumnDefinition(ByteBufferUtil.bytes("s1"));
private final ColumnDefinition s2 = metadata.getColumnDefinition(ByteBufferUtil.bytes("s2"));
private final ColumnDefinition v1 = metadata.getColumnDefinition(ByteBufferUtil.bytes("v1"));
private final ColumnDefinition v2 = metadata.getColumnDefinition(ByteBufferUtil.bytes("v2"));
private final CellPath path0 = CellPath.create(ByteBufferUtil.bytes(0));
private final CellPath path1 = CellPath.create(ByteBufferUtil.bytes(1));
private final CellPath path2 = CellPath.create(ByteBufferUtil.bytes(2));
private final CellPath path3 = CellPath.create(ByteBufferUtil.bytes(3));
private final CellPath path4 = CellPath.create(ByteBufferUtil.bytes(4));
@Parameterized.Parameter
public boolean anyNodeOn30;
@Parameterized.Parameters(name = "{index}: anyNodeOn30={0}")
public static Collection<Object[]> data()
{
CFMetaData metadata = CFMetaData.Builder.create("ks", "table")
.withPartitioner(Murmur3Partitioner.instance)
.addPartitionKey("pk", Int32Type.instance)
.addClusteringColumn("ck", Int32Type.instance)
.addRegularColumn("v1", Int32Type.instance)
.addRegularColumn("v2", Int32Type.instance)
.addRegularColumn("v3", Int32Type.instance)
.build();
ColumnDefinition v1 = metadata.getColumnDefinition(ByteBufferUtil.bytes("v1"));
testRoundTrip(ColumnFilter.all(metadata), metadata, MessagingService.VERSION_30);
testRoundTrip(ColumnFilter.all(metadata), metadata, MessagingService.VERSION_3014);
testRoundTrip(ColumnFilter.selection(metadata.partitionColumns().without(v1)), metadata, MessagingService.VERSION_30);
testRoundTrip(ColumnFilter.selection(metadata.partitionColumns().without(v1)), metadata, MessagingService.VERSION_3014);
return Arrays.asList(new Object[]{ true }, new Object[]{ false });
}
private static void testRoundTrip(ColumnFilter columnFilter, CFMetaData metadata, int version) throws Exception
@BeforeClass
public static void beforeClass()
{
DataOutputBuffer output = new DataOutputBuffer();
serializer.serialize(columnFilter, output, version);
Assert.assertEquals(serializer.serializedSize(columnFilter, version), output.position());
DataInputPlus input = new DataInputBuffer(output.buffer(), false);
Assert.assertEquals(serializer.deserialize(input, version, metadata), columnFilter);
DatabaseDescriptor.clientInitialization();
}
/**
* Tests whether a filter fetches and/or queries columns and cells.
*/
@Test
public void testFetchedQueried()
@Before
public void before()
{
CFMetaData metadata = CFMetaData.Builder.create("ks", "table")
.withPartitioner(Murmur3Partitioner.instance)
.addPartitionKey("k", Int32Type.instance)
.addRegularColumn("simple", Int32Type.instance)
.addRegularColumn("complex", SetType.getInstance(Int32Type.instance, true))
.build();
Gossiper.instance.setAnyNodeOn30(anyNodeOn30);
}
ColumnDefinition simple = metadata.getColumnDefinition(ByteBufferUtil.bytes("simple"));
ColumnDefinition complex = metadata.getColumnDefinition(ByteBufferUtil.bytes("complex"));
CellPath path1 = CellPath.create(ByteBufferUtil.bytes(1));
CellPath path2 = CellPath.create(ByteBufferUtil.bytes(2));
ColumnFilter filter;
// Select all
// select only the simple column, without table metadata
filter = ColumnFilter.selection(PartitionColumns.builder().add(simple).build());
assertFetchedQueried(true, true, filter, simple);
assertFetchedQueried(false, false, filter, complex);
assertFetchedQueried(false, false, filter, complex, path1);
assertFetchedQueried(false, false, filter, complex, path2);
@Test
public void testSelectAll()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("*/*", filter.toString());
assertFetchedQueried(true, true, filter, v1, v2, s1, s2);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
};
// select only the complex column, without table metadata
filter = ColumnFilter.selection(PartitionColumns.builder().add(complex).build());
assertFetchedQueried(false, false, filter, simple);
assertFetchedQueried(true, true, filter, complex);
assertFetchedQueried(true, true, filter, complex, path1);
assertFetchedQueried(true, true, filter, complex, path2);
check.accept(ColumnFilter.all(metadata));
check.accept(ColumnFilter.allColumnsBuilder(metadata).build());
}
// select both the simple and complex columns, without table metadata
filter = ColumnFilter.selection(PartitionColumns.builder().add(simple).add(complex).build());
assertFetchedQueried(true, true, filter, simple);
assertFetchedQueried(true, true, filter, complex);
assertFetchedQueried(true, true, filter, complex, path1);
assertFetchedQueried(true, true, filter, complex, path2);
// Selections
// select only the simple column, with table metadata
filter = ColumnFilter.selection(metadata, PartitionColumns.builder().add(simple).build());
assertFetchedQueried(true, true, filter, simple);
assertFetchedQueried(true, false, filter, complex);
assertFetchedQueried(true, false, filter, complex, path1);
assertFetchedQueried(true, false, filter, complex, path2);
@Test
public void testSelectNothing()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("[]", filter.toString());
assertFetchedQueried(false, false, filter, v1, v2, s1, s2);
assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
};
// select only the complex column, with table metadata
filter = ColumnFilter.selection(metadata, PartitionColumns.builder().add(complex).build());
assertFetchedQueried(true, false, filter, simple);
assertFetchedQueried(true, true, filter, complex);
assertFetchedQueried(true, true, filter, complex, path1);
assertFetchedQueried(true, true, filter, complex, path2);
check.accept(ColumnFilter.selection(PartitionColumns.NONE));
check.accept(ColumnFilter.selectionBuilder().build());
}
// select both the simple and complex columns, with table metadata
filter = ColumnFilter.selection(metadata, PartitionColumns.builder().add(simple).add(complex).build());
assertFetchedQueried(true, true, filter, simple);
assertFetchedQueried(true, true, filter, complex);
assertFetchedQueried(true, true, filter, complex, path1);
assertFetchedQueried(true, true, filter, complex, path2);
@Test
public void testSelectSimpleColumn()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("[v1]", filter.toString());
assertFetchedQueried(true, true, filter, v1);
assertFetchedQueried(false, false, filter, v2, s1, s2);
assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
};
// select only the simple column, with selection builder
filter = ColumnFilter.selectionBuilder().add(simple).build();
assertFetchedQueried(true, true, filter, simple);
assertFetchedQueried(false, false, filter, complex);
assertFetchedQueried(false, false, filter, complex, path1);
assertFetchedQueried(false, false, filter, complex, path2);
check.accept(ColumnFilter.selection(PartitionColumns.builder().add(v1).build()));
check.accept(ColumnFilter.selectionBuilder().add(v1).build());
}
// select only a cell of the complex column, with selection builder
filter = ColumnFilter.selectionBuilder().select(complex, path1).build();
assertFetchedQueried(false, false, filter, simple);
assertFetchedQueried(true, true, filter, complex);
assertFetchedQueried(true, true, filter, complex, path1);
assertFetchedQueried(true, false, filter, complex, path2);
@Test
public void testSelectComplexColumn()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("[v2]", filter.toString());
assertFetchedQueried(true, true, filter, v2);
assertFetchedQueried(false, false, filter, v1, s1, s2);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
};
// select both the simple column and a cell of the complex column, with selection builder
filter = ColumnFilter.selectionBuilder().add(simple).select(complex, path1).build();
assertFetchedQueried(true, true, filter, simple);
assertFetchedQueried(true, true, filter, complex);
assertFetchedQueried(true, true, filter, complex, path1);
assertFetchedQueried(true, false, filter, complex, path2);
check.accept(ColumnFilter.selection(PartitionColumns.builder().add(v2).build()));
check.accept(ColumnFilter.selectionBuilder().add(v2).build());
}
@Test
public void testSelectStaticColumn()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("[s1]", filter.toString());
assertFetchedQueried(true, true, filter, s1);
assertFetchedQueried(false, false, filter, v1, v2, s2);
assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
};
check.accept(ColumnFilter.selection(PartitionColumns.builder().add(s1).build()));
check.accept(ColumnFilter.selectionBuilder().add(s1).build());
}
@Test
public void testSelectStaticComplexColumn()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("[s2]", filter.toString());
assertFetchedQueried(true, true, filter, s2);
assertFetchedQueried(false, false, filter, v1, v2, s1);
assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
};
check.accept(ColumnFilter.selection(PartitionColumns.builder().add(s2).build()));
check.accept(ColumnFilter.selectionBuilder().add(s2).build());
}
@Test
public void testSelectColumns()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertEquals("[s1, s2, v1, v2]", filter.toString());
assertFetchedQueried(true, true, filter, v1, v2, s1, s2);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
};
check.accept(ColumnFilter.selection(PartitionColumns.builder().add(v1).add(v2).add(s1).add(s2).build()));
check.accept(ColumnFilter.selectionBuilder().add(v1).add(v2).add(s1).add(s2).build());
}
@Test
public void testSelectIndividualCells()
{
ColumnFilter filter = ColumnFilter.selectionBuilder().select(v2, path1).select(v2, path3).build();
testRoundTrips(filter);
assertEquals("[v2[1], v2[3]]", filter.toString());
assertFetchedQueried(true, true, filter, v2);
assertFetchedQueried(false, false, filter, v1, s1, s2);
assertCellFetchedQueried(true, true, filter, v2, path1, path3);
assertCellFetchedQueried(false, false, filter, v2, path0, path2, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
}
@Test
public void testSelectIndividualCellsFromStatic()
{
ColumnFilter filter = ColumnFilter.selectionBuilder().select(s2, path1).select(s2, path3).build();
testRoundTrips(filter);
assertEquals("[s2[1], s2[3]]", filter.toString());
assertFetchedQueried(true, true, filter, s2);
assertFetchedQueried(false, false, filter, v1, v2, s1);
assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path1, path3);
assertCellFetchedQueried(false, false, filter, s2, path0, path2, path4);
}
@Test
public void testSelectCellSlice()
{
ColumnFilter filter = ColumnFilter.selectionBuilder().slice(v2, path1, path3).build();
testRoundTrips(filter);
assertEquals("[v2[1:3]]", filter.toString());
assertFetchedQueried(true, true, filter, v2);
assertFetchedQueried(false, false, filter, v1, s1, s2);
assertCellFetchedQueried(true, true, filter, v2, path1, path2, path3);
assertCellFetchedQueried(false, false, filter, v2, path0, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
}
@Test
public void testSelectCellSliceFromStatic()
{
ColumnFilter filter = ColumnFilter.selectionBuilder().slice(s2, path1, path3).build();
testRoundTrips(filter);
assertEquals("[s2[1:3]]", filter.toString());
assertFetchedQueried(true, true, filter, s2);
assertFetchedQueried(false, false, filter, v1, v2, s1);
assertCellFetchedQueried(false, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path1, path2, path3);
assertCellFetchedQueried(false, false, filter, s2, path0, path4);
}
@Test
public void testSelectColumnsWithCellsAndSlices()
{
ColumnFilter filter = ColumnFilter.selectionBuilder()
.add(v1)
.add(s1)
.slice(v2, path0, path2)
.select(v2, path4)
.select(s2, path0)
.slice(s2, path2, path4)
.build();
testRoundTrips(filter);
assertEquals("[s1, s2[0], s2[2:4], v1, v2[0:2], v2[4]]", filter.toString());
assertFetchedQueried(true, true, filter, v1, v2, s1, s2);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path4);
assertCellFetchedQueried(false, false, filter, v2, path3);
assertCellFetchedQueried(true, true, filter, s2, path0, path2, path3, path4);
assertCellFetchedQueried(false, false, filter, s2, path1);
}
// select with metadata
@Test
public void testSelectSimpleColumnWithMetadata()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertFetchedQueried(true, true, filter, v1);
if (anyNodeOn30)
{
assertEquals("*/*", filter.toString());
assertFetchedQueried(true, true, filter, s1, s2, v2);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
}
else
{
assertEquals("*/[v1]", filter.toString());
assertFetchedQueried(true, false, filter, s1, s2, v2);
assertCellFetchedQueried(true, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, false, filter, s2, path0, path1, path2, path3, path4);
}
};
check.accept(ColumnFilter.selection(metadata, PartitionColumns.builder().add(v1).build()));
check.accept(ColumnFilter.allColumnsBuilder(metadata).add(v1).build());
}
@Test
public void testSelectStaticColumnWithMetadata()
{
Consumer<ColumnFilter> check = filter -> {
testRoundTrips(filter);
assertFetchedQueried(true, true, filter, s1);
if (anyNodeOn30)
{
assertEquals("*/*", filter.toString());
assertFetchedQueried(true, true, filter, v1, v2, s2);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
}
else
{
assertEquals("*/[s1]", filter.toString());
assertFetchedQueried(true, false, filter, v1, v2, s2);
assertCellFetchedQueried(true, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(false, false, filter, s2, path0, path1, path2, path3, path4);
}
};
check.accept(ColumnFilter.selection(metadata, PartitionColumns.builder().add(s1).build()));
check.accept(ColumnFilter.allColumnsBuilder(metadata).add(s1).build());
}
@Test
public void testSelectCellWithMetadata()
{
ColumnFilter filter = ColumnFilter.allColumnsBuilder(metadata).select(v2, path1).build();
testRoundTrips(filter);
assertFetchedQueried(true, true, filter, v2);
if (anyNodeOn30)
{
assertEquals("*/*", filter.toString());
assertFetchedQueried(true, true, filter, s1, s2, v1);
assertCellFetchedQueried(true, true, filter, v2, path1);
assertCellFetchedQueried(true, false, filter, v2, path0, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path0, path1, path2, path3, path4);
}
else
{
assertEquals("*/[v2[1]]", filter.toString());
assertFetchedQueried(true, false, filter, s1, s2, v1);
assertCellFetchedQueried(true, true, filter, v2, path1);
assertCellFetchedQueried(true, false, filter, v2, path0, path2, path3, path4);
assertCellFetchedQueried(true, false, filter, s2, path0, path1, path2, path3, path4);
}
}
@Test
public void testSelectStaticColumnCellWithMetadata()
{
ColumnFilter filter = ColumnFilter.allColumnsBuilder(metadata).select(s2, path1).build();
testRoundTrips(filter);
assertFetchedQueried(true, true, filter, s2);
if (anyNodeOn30)
{
assertEquals("*/*", filter.toString());
assertFetchedQueried(true, true, filter, v1, v2, s1);
assertCellFetchedQueried(true, true, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path1);
assertCellFetchedQueried(true, false, filter, s2, path0, path2, path3, path4);
}
else
{
assertEquals("*/[s2[1]]", filter.toString());
assertFetchedQueried(true, false, filter, v1, v2, s1);
assertCellFetchedQueried(true, false, filter, v2, path0, path1, path2, path3, path4);
assertCellFetchedQueried(true, true, filter, s2, path1);
assertCellFetchedQueried(true, false, filter, s2, path0, path2, path3, path4);
}
}
private void testRoundTrips(ColumnFilter cf)
{
testRoundTrip(cf, MessagingService.VERSION_30);
testRoundTrip(cf, MessagingService.VERSION_3014);
}
private void testRoundTrip(ColumnFilter columnFilter, int version)
{
try
{
DataOutputBuffer output = new DataOutputBuffer();
serializer.serialize(columnFilter, output, version);
Assert.assertEquals(serializer.serializedSize(columnFilter, version), output.position());
DataInputPlus input = new DataInputBuffer(output.buffer(), false);
ColumnFilter deserialized = serializer.deserialize(input, version, metadata);
Assert.assertEquals(deserialized, columnFilter);
}
catch (IOException e)
{
throw Throwables.propagate(e);
}
}
private static void assertFetchedQueried(boolean expectedFetched,
boolean expectedQueried,
ColumnFilter filter,
ColumnDefinition column)
ColumnDefinition... columns)
{
assert !expectedQueried || expectedFetched;
boolean actualFetched = filter.fetches(column);
assertEquals(expectedFetched, actualFetched);
assertEquals(expectedQueried, actualFetched && filter.fetchedColumnIsQueried(column));
for (ColumnDefinition column : columns)
{
assertEquals(String.format("Expected fetches(%s) to be %s", column.name, expectedFetched),
expectedFetched, filter.fetches(column));
if (expectedFetched)
assertEquals(String.format("Expected fetchedColumnIsQueried(%s) to be %s", column.name, expectedQueried),
expectedQueried, filter.fetchedColumnIsQueried(column));
}
}
private static void assertFetchedQueried(boolean expectedFetched,
boolean expectedQueried,
ColumnFilter filter,
ColumnDefinition column,
CellPath path)
private static void assertCellFetchedQueried(boolean expectedFetched,
boolean expectedQueried,
ColumnFilter filter,
ColumnDefinition column,
CellPath... paths)
{
assert !expectedQueried || expectedFetched;
boolean actualFetched = filter.fetches(column);
assertEquals(expectedFetched, actualFetched);
assertEquals(expectedQueried, actualFetched && filter.fetchedCellIsQueried(column, path));
ColumnFilter.Tester tester = filter.newTester(column);
for (CellPath path : paths)
{
int p = ByteBufferUtil.toInt(path.get(0));
if (expectedFetched)
assertEquals(String.format("Expected fetchedCellIsQueried(%s:%s) to be %s", column.name, p, expectedQueried),
expectedQueried, filter.fetchedCellIsQueried(column, path));
if (tester != null)
{
assertEquals(String.format("Expected tester.fetches(%s:%s) to be %s", column.name, p, expectedFetched),
expectedFetched, tester.fetches(path));
if (expectedFetched)
assertEquals(String.format("Expected tester.fetchedCellIsQueried(%s:%s) to be %s", column.name, p, expectedQueried),
expectedQueried, tester.fetchedCellIsQueried(path));
}
}
}
}
}