mirror of https://github.com/apache/cassandra
1726 lines
78 KiB
Java
1726 lines
78 KiB
Java
/*
|
|
* 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.db;
|
|
|
|
import java.io.IOException;
|
|
import java.io.OutputStream;
|
|
import java.net.UnknownHostException;
|
|
import java.nio.ByteBuffer;
|
|
import java.util.ArrayList;
|
|
import java.util.Collections;
|
|
import java.util.EnumSet;
|
|
import java.util.HashMap;
|
|
import java.util.HashSet;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Set;
|
|
import java.util.function.Supplier;
|
|
import java.util.stream.IntStream;
|
|
|
|
import com.google.common.collect.ImmutableSet;
|
|
import com.google.common.collect.Lists;
|
|
import com.google.common.collect.Sets;
|
|
|
|
import org.junit.Assert;
|
|
import org.junit.BeforeClass;
|
|
import org.junit.Test;
|
|
|
|
import org.apache.cassandra.SchemaLoader;
|
|
import org.apache.cassandra.Util;
|
|
import org.apache.cassandra.config.Config;
|
|
import org.apache.cassandra.config.DatabaseDescriptor;
|
|
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
|
|
import org.apache.cassandra.db.filter.ColumnFilter;
|
|
import org.apache.cassandra.db.filter.DataLimits;
|
|
import org.apache.cassandra.db.filter.RowFilter;
|
|
import org.apache.cassandra.db.marshal.AsciiType;
|
|
import org.apache.cassandra.db.marshal.BytesType;
|
|
import org.apache.cassandra.db.marshal.CounterColumnType;
|
|
import org.apache.cassandra.db.marshal.Int32Type;
|
|
import org.apache.cassandra.db.marshal.ReversedType;
|
|
import org.apache.cassandra.db.marshal.SetType;
|
|
import org.apache.cassandra.db.partitions.FilteredPartition;
|
|
import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
|
|
import org.apache.cassandra.db.partitions.Partition;
|
|
import org.apache.cassandra.db.partitions.PartitionIterator;
|
|
import org.apache.cassandra.db.partitions.PartitionUpdate;
|
|
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
|
|
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
|
|
import org.apache.cassandra.db.rows.DeserializationHelper;
|
|
import org.apache.cassandra.db.rows.Row;
|
|
import org.apache.cassandra.db.rows.RowIterator;
|
|
import org.apache.cassandra.db.rows.Unfiltered;
|
|
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
|
|
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
|
|
import org.apache.cassandra.dht.Range;
|
|
import org.apache.cassandra.dht.Token;
|
|
import org.apache.cassandra.exceptions.ConfigurationException;
|
|
import org.apache.cassandra.exceptions.QueryCancelledException;
|
|
import org.apache.cassandra.io.sstable.format.SSTableReader;
|
|
import org.apache.cassandra.io.util.DataInputBuffer;
|
|
import org.apache.cassandra.io.util.DataOutputBuffer;
|
|
import org.apache.cassandra.io.util.WrappedDataOutputStreamPlus;
|
|
import org.apache.cassandra.locator.InetAddressAndPort;
|
|
import org.apache.cassandra.metrics.ClearableHistogram;
|
|
import org.apache.cassandra.net.Message;
|
|
import org.apache.cassandra.net.MessagingService;
|
|
import org.apache.cassandra.net.Verb;
|
|
import org.apache.cassandra.repair.consistent.LocalSessionAccessor;
|
|
import org.apache.cassandra.schema.CachingParams;
|
|
import org.apache.cassandra.schema.KeyspaceMetadata;
|
|
import org.apache.cassandra.schema.KeyspaceParams;
|
|
import org.apache.cassandra.schema.Schema;
|
|
import org.apache.cassandra.schema.SchemaTestUtil;
|
|
import org.apache.cassandra.schema.TableMetadata;
|
|
import org.apache.cassandra.schema.TableParams;
|
|
import org.apache.cassandra.service.ActiveRepairService;
|
|
import org.apache.cassandra.streaming.PreviewKind;
|
|
import org.apache.cassandra.tracing.Tracing;
|
|
import org.apache.cassandra.utils.ByteBufferUtil;
|
|
import org.apache.cassandra.utils.FBUtilities;
|
|
import org.apache.cassandra.utils.TimeUUID;
|
|
|
|
import static org.apache.cassandra.utils.ByteBufferUtil.EMPTY_BYTE_BUFFER;
|
|
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
|
|
import static org.junit.Assert.assertEquals;
|
|
import static org.junit.Assert.assertFalse;
|
|
import static org.junit.Assert.assertNotEquals;
|
|
import static org.junit.Assert.assertTrue;
|
|
import static org.junit.Assert.fail;
|
|
|
|
public class ReadCommandTest
|
|
{
|
|
private static final String CREATE = "1";
|
|
private static final String DELETE = "-1";
|
|
|
|
private static final String KEYSPACE = "ReadCommandTest";
|
|
private static final String CF1 = "Standard1";
|
|
private static final String CF2 = "Standard2";
|
|
private static final String CF3 = "Standard3";
|
|
private static final String CF4 = "Standard4";
|
|
private static final String CF5 = "Standard5";
|
|
private static final String CF6 = "Standard6";
|
|
private static final String CF7 = "Counter7";
|
|
private static final String CF8 = "Standard8";
|
|
private static final String CF9 = "Standard9";
|
|
private static final String CF10 = "Standard10";
|
|
private static final String CF11 = "Standard11";
|
|
private static final String CF12 = "Standard12";
|
|
private static final String CF13 = "Standard13";
|
|
private static final String CF14 = "Standard14";
|
|
|
|
|
|
private static final InetAddressAndPort REPAIR_COORDINATOR;
|
|
static {
|
|
try
|
|
{
|
|
REPAIR_COORDINATOR = InetAddressAndPort.getByName("10.0.0.1");
|
|
}
|
|
catch (UnknownHostException e)
|
|
{
|
|
|
|
throw new AssertionError(e);
|
|
}
|
|
}
|
|
|
|
@BeforeClass
|
|
public static void defineSchema() throws ConfigurationException
|
|
{
|
|
DatabaseDescriptor.daemonInitialization();
|
|
|
|
TableMetadata.Builder metadata1 = SchemaLoader.standardCFMD(KEYSPACE, CF1);
|
|
|
|
TableMetadata.Builder metadata2 =
|
|
TableMetadata.builder(KEYSPACE, CF2)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata3 =
|
|
TableMetadata.builder(KEYSPACE, CF3)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata4 =
|
|
TableMetadata.builder(KEYSPACE, CF4)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata5 =
|
|
TableMetadata.builder(KEYSPACE, CF5)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata6 =
|
|
TableMetadata.builder(KEYSPACE, CF6)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addStaticColumn("s", AsciiType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.caching(CachingParams.CACHE_EVERYTHING);
|
|
|
|
TableMetadata.Builder metadata7 =
|
|
TableMetadata.builder(KEYSPACE, CF7)
|
|
.flags(EnumSet.of(TableMetadata.Flag.COUNTER, TableMetadata.Flag.COMPOUND))
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("c", CounterColumnType.instance);
|
|
|
|
TableMetadata.Builder metadata8 =
|
|
TableMetadata.builder(KEYSPACE, CF8)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", SetType.getInstance(AsciiType.instance, true));
|
|
|
|
TableMetadata.Builder metadata9 =
|
|
TableMetadata.builder(KEYSPACE, CF9)
|
|
.addPartitionKeyColumn("key", Int32Type.instance)
|
|
.addClusteringColumn("col", ReversedType.getInstance(Int32Type.instance))
|
|
.addRegularColumn("a", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata10 =
|
|
TableMetadata.builder(KEYSPACE, CF10)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata11 =
|
|
TableMetadata.builder(KEYSPACE, CF11)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata12 =
|
|
TableMetadata.builder(KEYSPACE, CF12)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata13 =
|
|
TableMetadata.builder(KEYSPACE, CF13)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
TableMetadata.Builder metadata14 =
|
|
TableMetadata.builder(KEYSPACE, CF14)
|
|
.addPartitionKeyColumn("key", BytesType.instance)
|
|
.addClusteringColumn("col", AsciiType.instance)
|
|
.addRegularColumn("a", AsciiType.instance)
|
|
.addRegularColumn("b", AsciiType.instance)
|
|
.addRegularColumn("c", AsciiType.instance)
|
|
.addRegularColumn("d", AsciiType.instance)
|
|
.addRegularColumn("e", AsciiType.instance)
|
|
.addRegularColumn("f", AsciiType.instance);
|
|
|
|
SchemaLoader.prepareServer();
|
|
SchemaLoader.createKeyspace(KEYSPACE,
|
|
KeyspaceParams.simple(1),
|
|
metadata1,
|
|
metadata2,
|
|
metadata3,
|
|
metadata4,
|
|
metadata5,
|
|
metadata6,
|
|
metadata7,
|
|
metadata8,
|
|
metadata9,
|
|
metadata10,
|
|
metadata11,
|
|
metadata12,
|
|
metadata13,
|
|
metadata14);
|
|
|
|
LocalSessionAccessor.startup();
|
|
}
|
|
|
|
@Test
|
|
public void testPartitionRangeAbort() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF1);
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key1"))
|
|
.clustering("Column1")
|
|
.add("val", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key2"))
|
|
.clustering("Column1")
|
|
.add("val", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs).build();
|
|
assertEquals(2, Util.getAll(readCommand).size());
|
|
|
|
readCommand.abort();
|
|
boolean cancelled = false;
|
|
try
|
|
{
|
|
Util.getAll(readCommand);
|
|
}
|
|
catch (QueryCancelledException e)
|
|
{
|
|
cancelled = true;
|
|
}
|
|
assertTrue(cancelled);
|
|
}
|
|
|
|
@Test
|
|
public void testSinglePartitionSliceAbort()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
|
|
cfs.truncateBlocking();
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("cc")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("dd")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build();
|
|
|
|
List<FilteredPartition> partitions = Util.getAll(readCommand);
|
|
assertEquals(1, partitions.size());
|
|
assertEquals(2, partitions.get(0).rowCount());
|
|
|
|
readCommand.abort();
|
|
boolean cancelled = false;
|
|
try
|
|
{
|
|
Util.getAll(readCommand);
|
|
}
|
|
catch (QueryCancelledException e)
|
|
{
|
|
cancelled = true;
|
|
}
|
|
assertTrue(cancelled);
|
|
}
|
|
|
|
@Test
|
|
public void testSinglePartitionNamesAbort()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
|
|
cfs.truncateBlocking();
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("cc")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("dd")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("cc").includeRow("dd").build();
|
|
|
|
List<FilteredPartition> partitions = Util.getAll(readCommand);
|
|
assertEquals(1, partitions.size());
|
|
assertEquals(2, partitions.get(0).rowCount());
|
|
|
|
readCommand.abort();
|
|
boolean cancelled = false;
|
|
try
|
|
{
|
|
Util.getAll(readCommand);
|
|
}
|
|
catch (QueryCancelledException e)
|
|
{
|
|
cancelled = true;
|
|
}
|
|
assertTrue(cancelled);
|
|
}
|
|
|
|
@Test
|
|
public void testSinglePartitionGroupMerge() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF3);
|
|
|
|
String[][][] groups = new String[][][] {
|
|
new String[][] {
|
|
new String[] { CREATE, "key1", "aa", "a" },
|
|
new String[] { CREATE, "key2", "bb", "b" },
|
|
new String[] { CREATE, "key3", "cc", "c" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key3", "dd", "d" },
|
|
new String[] { CREATE, "key2", "ee", "e" },
|
|
new String[] { CREATE, "key1", "ff", "f" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key6", "aa", "a" },
|
|
new String[] { CREATE, "key5", "bb", "b" },
|
|
new String[] { CREATE, "key4", "cc", "c" }
|
|
},
|
|
new String[][] {
|
|
new String[] { DELETE, "key6", "aa", "a" },
|
|
new String[] { DELETE, "key2", "bb", "b" }
|
|
}
|
|
};
|
|
|
|
// Given the data above, when the keys are sorted and the deletions removed, we should
|
|
// get these clustering rows in this order
|
|
String[] expectedRows = new String[] { "aa", "ff", "ee", "cc", "dd", "cc", "bb"};
|
|
|
|
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
|
|
long nowInSeconds = FBUtilities.nowInSeconds();
|
|
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
|
|
RowFilter rowFilter = RowFilter.create(true);
|
|
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
|
|
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(cfs.metadata().comparator, slice), false);
|
|
|
|
for (String[][] group : groups)
|
|
{
|
|
cfs.truncateBlocking();
|
|
|
|
List<SinglePartitionReadCommand> commands = new ArrayList<>(group.length);
|
|
|
|
for (String[] data : group)
|
|
{
|
|
if (data[0].equals(CREATE))
|
|
{
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(data[1]))
|
|
.clustering(data[2])
|
|
.add(data[3], ByteBufferUtil.bytes("blah"))
|
|
.build()
|
|
.apply();
|
|
}
|
|
else
|
|
{
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), FBUtilities.timestampMicros(), ByteBufferUtil.bytes(data[1]), data[2]).apply();
|
|
}
|
|
commands.add(SinglePartitionReadCommand.create(cfs.metadata(), nowInSeconds, columnFilter, rowFilter, DataLimits.NONE, Util.dk(data[1]), sliceFilter));
|
|
}
|
|
|
|
Util.flush(cfs);
|
|
|
|
ReadQuery query = SinglePartitionReadCommand.Group.create(commands, DataLimits.NONE);
|
|
|
|
try (ReadExecutionController executionController = query.executionController();
|
|
UnfilteredPartitionIterator iter = query.executeLocally(executionController);
|
|
DataOutputBuffer buffer = new DataOutputBuffer())
|
|
{
|
|
UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter,
|
|
columnFilter,
|
|
buffer,
|
|
MessagingService.current_version);
|
|
buffers.add(buffer.buffer());
|
|
}
|
|
}
|
|
|
|
// deserialize, merge and check the results are all there
|
|
List<UnfilteredPartitionIterator> iterators = new ArrayList<>();
|
|
|
|
for (ByteBuffer buffer : buffers)
|
|
{
|
|
try (DataInputBuffer in = new DataInputBuffer(buffer, true))
|
|
{
|
|
iterators.add(UnfilteredPartitionIterators.serializerForIntraNode().deserialize(in,
|
|
MessagingService.current_version,
|
|
cfs.metadata(),
|
|
columnFilter,
|
|
DeserializationHelper.Flag.LOCAL));
|
|
}
|
|
}
|
|
|
|
UnfilteredPartitionIterators.MergeListener listener =
|
|
new UnfilteredPartitionIterators.MergeListener()
|
|
{
|
|
public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
public void close()
|
|
{
|
|
|
|
}
|
|
};
|
|
|
|
try (PartitionIterator partitionIterator = UnfilteredPartitionIterators.filter(UnfilteredPartitionIterators.merge(iterators, listener), nowInSeconds))
|
|
{
|
|
|
|
int i = 0;
|
|
int numPartitions = 0;
|
|
while (partitionIterator.hasNext())
|
|
{
|
|
numPartitions++;
|
|
try(RowIterator rowIterator = partitionIterator.next())
|
|
{
|
|
while (rowIterator.hasNext())
|
|
{
|
|
Row row = rowIterator.next();
|
|
assertEquals("col=" + expectedRows[i++], row.clustering().toString(cfs.metadata()));
|
|
//System.out.print(row.toString(cfs.metadata, true));
|
|
}
|
|
}
|
|
}
|
|
|
|
assertEquals(5, numPartitions);
|
|
assertEquals(expectedRows.length, i);
|
|
}
|
|
}
|
|
|
|
@Test
|
|
public void testSerializer() throws IOException
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
|
|
new RowUpdateBuilder(cfs.metadata.get(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("dd")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("dd").build();
|
|
int messagingVersion = MessagingService.current_version;
|
|
FakeOutputStream out = new FakeOutputStream();
|
|
Tracing.instance.newSession(Tracing.TraceType.QUERY);
|
|
Message<ReadCommand> messageOut = Message.out(Verb.READ_REQ, readCommand);
|
|
long size = messageOut.serializedSize(messagingVersion);
|
|
Message.serializer.serialize(messageOut, new WrappedDataOutputStreamPlus(out), messagingVersion);
|
|
Assert.assertEquals(size, out.count);
|
|
}
|
|
|
|
static class FakeOutputStream extends OutputStream
|
|
{
|
|
long count;
|
|
|
|
public void write(int b) throws IOException
|
|
{
|
|
count++;
|
|
}
|
|
}
|
|
|
|
@Test
|
|
public void testCountDeletedRows() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF4);
|
|
|
|
String[][][] groups = new String[][][] {
|
|
new String[][] {
|
|
new String[] { CREATE, "key1", "aa", "a" },
|
|
new String[] { CREATE, "key2", "bb", "b" },
|
|
new String[] { CREATE, "key3", "cc", "c" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key3", "dd", "d" },
|
|
new String[] { CREATE, "key2", "ee", "e" },
|
|
new String[] { CREATE, "key1", "ff", "f" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key6", "aa", "a" },
|
|
new String[] { CREATE, "key5", "bb", "b" },
|
|
new String[] { CREATE, "key4", "cc", "c" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key2", "aa", "a" },
|
|
new String[] { CREATE, "key2", "cc", "c" },
|
|
new String[] { CREATE, "key2", "dd", "d" }
|
|
},
|
|
new String[][] {
|
|
new String[] { DELETE, "key6", "aa", "a" },
|
|
new String[] { DELETE, "key2", "bb", "b" },
|
|
new String[] { DELETE, "key2", "ee", "e" },
|
|
new String[] { DELETE, "key2", "aa", "a" },
|
|
new String[] { DELETE, "key2", "cc", "c" },
|
|
new String[] { DELETE, "key2", "dd", "d" }
|
|
}
|
|
};
|
|
|
|
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
|
|
long nowInSeconds = FBUtilities.nowInSeconds();
|
|
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
|
|
RowFilter rowFilter = RowFilter.create(true);
|
|
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
|
|
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(
|
|
Slices.with(cfs.metadata().comparator, slice), false);
|
|
|
|
for (String[][] group : groups)
|
|
{
|
|
cfs.truncateBlocking();
|
|
|
|
List<SinglePartitionReadCommand> commands = new ArrayList<>(group.length);
|
|
|
|
for (String[] data : group)
|
|
{
|
|
if (data[0].equals(CREATE))
|
|
{
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(data[1]))
|
|
.clustering(data[2])
|
|
.add(data[3], ByteBufferUtil.bytes("blah"))
|
|
.build()
|
|
.apply();
|
|
}
|
|
else
|
|
{
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), FBUtilities.timestampMicros(),
|
|
ByteBufferUtil.bytes(data[1]), data[2]).apply();
|
|
}
|
|
commands.add(SinglePartitionReadCommand.create(cfs.metadata(), nowInSeconds, columnFilter, rowFilter,
|
|
DataLimits.NONE, Util.dk(data[1]), sliceFilter));
|
|
}
|
|
|
|
Util.flush(cfs);
|
|
|
|
ReadQuery query = SinglePartitionReadCommand.Group.create(commands, DataLimits.NONE);
|
|
|
|
try (ReadExecutionController executionController = query.executionController();
|
|
UnfilteredPartitionIterator iter = query.executeLocally(executionController);
|
|
DataOutputBuffer buffer = new DataOutputBuffer())
|
|
{
|
|
UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter,
|
|
columnFilter,
|
|
buffer,
|
|
MessagingService.current_version);
|
|
buffers.add(buffer.buffer());
|
|
}
|
|
}
|
|
|
|
assertEquals(5, cfs.metric.tombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
}
|
|
|
|
@Test
|
|
public void testCountWithNoDeletedRow() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF5);
|
|
|
|
String[][][] groups = new String[][][] {
|
|
new String[][] {
|
|
new String[] { CREATE, "key1", "aa", "a" },
|
|
new String[] { CREATE, "key2", "bb", "b" },
|
|
new String[] { CREATE, "key3", "cc", "c" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key3", "dd", "d" },
|
|
new String[] { CREATE, "key2", "ee", "e" },
|
|
new String[] { CREATE, "key1", "ff", "f" }
|
|
},
|
|
new String[][] {
|
|
new String[] { CREATE, "key6", "aa", "a" },
|
|
new String[] { CREATE, "key5", "bb", "b" },
|
|
new String[] { CREATE, "key4", "cc", "c" }
|
|
}
|
|
};
|
|
|
|
List<ByteBuffer> buffers = new ArrayList<>(groups.length);
|
|
long nowInSeconds = FBUtilities.nowInSeconds();
|
|
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
|
|
RowFilter rowFilter = RowFilter.create(true);
|
|
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
|
|
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(
|
|
Slices.with(cfs.metadata().comparator, slice), false);
|
|
|
|
for (String[][] group : groups)
|
|
{
|
|
cfs.truncateBlocking();
|
|
|
|
List<SinglePartitionReadCommand> commands = new ArrayList<>(group.length);
|
|
|
|
for (String[] data : group)
|
|
{
|
|
if (data[0].equals(CREATE))
|
|
{
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(data[1]))
|
|
.clustering(data[2])
|
|
.add(data[3], ByteBufferUtil.bytes("blah"))
|
|
.build()
|
|
.apply();
|
|
}
|
|
else
|
|
{
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), FBUtilities.timestampMicros(),
|
|
ByteBufferUtil.bytes(data[1]), data[2]).apply();
|
|
}
|
|
commands.add(SinglePartitionReadCommand.create(cfs.metadata(), nowInSeconds, columnFilter, rowFilter,
|
|
DataLimits.NONE, Util.dk(data[1]), sliceFilter));
|
|
}
|
|
|
|
Util.flush(cfs);
|
|
|
|
ReadQuery query = SinglePartitionReadCommand.Group.create(commands, DataLimits.NONE);
|
|
|
|
try (ReadExecutionController executionController = query.executionController();
|
|
UnfilteredPartitionIterator iter = query.executeLocally(executionController);
|
|
DataOutputBuffer buffer = new DataOutputBuffer())
|
|
{
|
|
UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter,
|
|
columnFilter,
|
|
buffer,
|
|
MessagingService.current_version);
|
|
buffers.add(buffer.buffer());
|
|
}
|
|
}
|
|
|
|
assertEquals(1, cfs.metric.tombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
}
|
|
|
|
@Test
|
|
public void testCountPurgeableRowTombstones() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF10);
|
|
TestWriteOperation[] operations = new TestWriteOperation[]
|
|
{
|
|
TestWriteOperation.insert("key1", "aa", "a"),
|
|
TestWriteOperation.insert("key1", "ff", "f"),
|
|
|
|
TestWriteOperation.insert("key2", "aa", "e"),
|
|
TestWriteOperation.deleteRow("key2", "aa", PURGEABLE_DELETION),
|
|
TestWriteOperation.deleteRow("key2", "bb", NEW_DELETION),
|
|
TestWriteOperation.deleteRow("key2", "cc", PURGEABLE_DELETION),
|
|
TestWriteOperation.deleteRow("key2", "dd", PURGEABLE_DELETION),
|
|
TestWriteOperation.deleteRow("key2", "ee", NEW_DELETION),
|
|
};
|
|
|
|
runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row);
|
|
|
|
assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount());
|
|
assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin());
|
|
assertEquals(3, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
}
|
|
@Test
|
|
public void testCountPurgeablePartitionTombstones() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF11);
|
|
TestWriteOperation[] operations = new TestWriteOperation[]
|
|
{
|
|
TestWriteOperation.insert("key1", "aa", "a"),
|
|
TestWriteOperation.insert("key1", "ff", "f"),
|
|
|
|
TestWriteOperation.insert("key2", "aa", "a"),
|
|
TestWriteOperation.insert("key2", "cc", "c"),
|
|
TestWriteOperation.insert("key2", "dd", "d"),
|
|
|
|
TestWriteOperation.deletePartition("key2", PURGEABLE_DELETION),
|
|
TestWriteOperation.deletePartition("key3", NEW_DELETION)
|
|
};
|
|
runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row);
|
|
|
|
assertEquals(3, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount());
|
|
assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin());
|
|
assertEquals(1, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
}
|
|
|
|
@Test
|
|
public void testCountPurgeableCellTombstones() throws Exception
|
|
{
|
|
DatabaseDescriptor.setPurgeableTobmstonesMetricGranularity(Config.TombstonesMetricGranularity.cell);
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF12);
|
|
TestWriteOperation[] operations = new TestWriteOperation[]
|
|
{
|
|
TestWriteOperation.insert("key1", "aa", "a"),
|
|
TestWriteOperation.insert("key1", "ff", "f"),
|
|
|
|
TestWriteOperation.insert("key2", "aa", "a"),
|
|
TestWriteOperation.deleteCell("key2", "aa", "b", PURGEABLE_DELETION),
|
|
TestWriteOperation.deleteCell("key2", "aa", "f", NEW_DELETION),
|
|
TestWriteOperation.insert("key2", "cc", "c"),
|
|
TestWriteOperation.insert("key2", "dd", "d")
|
|
};
|
|
runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.cell);
|
|
|
|
assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount());
|
|
assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin());
|
|
assertEquals(1, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
|
|
}
|
|
|
|
/**
|
|
* Test purgeable tombstones count for range tombstones with non-overlapping ranges,
|
|
* i.e. only Bound (not Boundary) Markers will be created and counted
|
|
*/
|
|
@Test
|
|
public void testCountPurgeableRangeTombstones_nonOverlappingRanges() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF13);
|
|
TestWriteOperation[] operations = new TestWriteOperation[]
|
|
{
|
|
TestWriteOperation.insert("key1", "aa", "a"),
|
|
TestWriteOperation.insert("key1", "ff", "f"),
|
|
|
|
TestWriteOperation.insert("key2", "aa", "a"),
|
|
TestWriteOperation.insert("key2", "cc", "c"),
|
|
TestWriteOperation.insert("key2", "dd", "d"),
|
|
|
|
TestWriteOperation.deleteRange("key2", "aa", "bb", NEW_DELETION),
|
|
TestWriteOperation.deleteRange("key2", "dd", "ee", PURGEABLE_DELETION),
|
|
TestWriteOperation.deleteRange("key2", "ff", "ff", PURGEABLE_DELETION)
|
|
};
|
|
runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row);
|
|
|
|
assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getCount());
|
|
assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin());
|
|
assertEquals(4, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
}
|
|
|
|
/**
|
|
* Test purgeable tombstones count for range tombstones with overlapping ranges
|
|
*/
|
|
@Test
|
|
public void testCountPurgeableRangeTombstones_overlappingRanges() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF14);
|
|
|
|
TestWriteOperation[] operations = new TestWriteOperation[]
|
|
{
|
|
TestWriteOperation.insert("key1", "aa", "a"),
|
|
TestWriteOperation.insert("key1", "ff", "f"),
|
|
|
|
TestWriteOperation.insert("key2", "aa", "a"),
|
|
TestWriteOperation.insert("key2", "bb", "b"),
|
|
TestWriteOperation.insert("key2", "cc", "c"),
|
|
TestWriteOperation.insert("key2", "dd", "d"),
|
|
TestWriteOperation.insert("key2", "ee", "e"),
|
|
|
|
// this range tombstone is non-purgeable and overlaps with the next one,
|
|
// so it will create one non-purgeable bound marker
|
|
// and one non-purgeable boundary marker so TWO NON-PURGEABLE tombstones
|
|
TestWriteOperation.deleteRange("key2", "aa", "bb", NEW_DELETION),
|
|
|
|
// this range tombstone is purgeable and overlaps with previous and next ones,
|
|
// so it will create one non-purgeable bound marker
|
|
// and one non-purgeable boundary marker so TWO non-purgeable tombstones will be counted
|
|
TestWriteOperation.deleteRange("key2", "bb", "ee", PURGEABLE_DELETION),
|
|
|
|
// this range tombstone is purgeable and overlaps with previous one,
|
|
// it has a different deletion time to not combine into a single range,
|
|
// so it will create one non-purgeable boundary marker (same as previous one)
|
|
// and one purgeable bound marker, so it will increment purgeable tombstones counter on one,
|
|
// we expect TWO purgeable tombstones in total
|
|
TestWriteOperation.deleteRange("key2", "ee", "ff", PURGEABLE_DELETION - 1)
|
|
};
|
|
|
|
runTestWriteOperationsAndReadResults(cfs, operations, Config.TombstonesMetricGranularity.row);
|
|
|
|
assertEquals(0, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMin());
|
|
assertEquals(2, cfs.metric.purgeableTombstoneScannedHistogram.cf.getSnapshot().getMax());
|
|
}
|
|
|
|
|
|
private static void runTestWriteOperationsAndReadResults(ColumnFamilyStore cfs, TestWriteOperation[] operations, Config.TombstonesMetricGranularity granularity) throws IOException
|
|
{
|
|
Config.TombstonesMetricGranularity original = DatabaseDescriptor.getPurgeableTobmstonesMetricGranularity();
|
|
DatabaseDescriptor.setPurgeableTobmstonesMetricGranularity(granularity);
|
|
try
|
|
{
|
|
Set<String> usedPartitionKeys = runWriteOperations(cfs, operations);
|
|
runPartitionReadCommands(cfs, usedPartitionKeys);
|
|
}
|
|
finally
|
|
{
|
|
DatabaseDescriptor.setPurgeableTobmstonesMetricGranularity(original);
|
|
cfs.truncateBlocking();
|
|
}
|
|
}
|
|
|
|
private static void runPartitionReadCommands(ColumnFamilyStore cfs, Set<String> partitionKeys) throws IOException
|
|
{
|
|
List<SinglePartitionReadCommand> commands = new ArrayList<>(partitionKeys.size());
|
|
long nowInSeconds = FBUtilities.nowInSeconds(); // all reads within a group must have the same nowInSec
|
|
for (String partitionKey : partitionKeys)
|
|
{
|
|
commands.add(getWholePartitionReadCommand(cfs, partitionKey, nowInSeconds));
|
|
}
|
|
executeReadCommands(commands);
|
|
}
|
|
|
|
private static Set<String> runWriteOperations(ColumnFamilyStore cfs, TestWriteOperation[] operations)
|
|
{
|
|
Set<String> usedPartitionKeys = new HashSet<>();
|
|
for (TestWriteOperation operation : operations)
|
|
{
|
|
if (operation.type == OperationType.CREATE)
|
|
{
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes(operation.partitionKey))
|
|
.clustering(operation.clusteringKey)
|
|
.add(operation.columnName, ByteBufferUtil.bytes(operation.columnValue))
|
|
.build()
|
|
.apply();
|
|
}
|
|
else if (operation.type == OperationType.DELETE_PARTITION)
|
|
{
|
|
new Mutation(PartitionUpdate.simpleBuilder(cfs.metadata(), ByteBufferUtil.bytes(operation.partitionKey))
|
|
.nowInSec(operation.deletionTime)
|
|
.delete()
|
|
.build()).apply();
|
|
}
|
|
else if (operation.type == OperationType.DELETE_RANGE)
|
|
{
|
|
new RowUpdateBuilder(cfs.metadata(), operation.deletionTime, 0L, ByteBufferUtil.bytes(operation.partitionKey))
|
|
.addRangeTombstone(operation.clusteringRangeStart, operation.clusteringRangeEnd).build().apply();
|
|
}
|
|
else if (operation.type == OperationType.DELETE_ROW)
|
|
{
|
|
RowUpdateBuilder.deleteRowAt(cfs.metadata(), 0, operation.deletionTime,
|
|
ByteBufferUtil.bytes(operation.partitionKey), operation.clusteringKey
|
|
).apply();
|
|
}
|
|
else if (operation.type == OperationType.DELETE_CELL)
|
|
{
|
|
new RowUpdateBuilder(cfs.metadata(), operation.deletionTime, 0L, ByteBufferUtil.bytes(operation.partitionKey))
|
|
.clustering(operation.clusteringKey)
|
|
.delete(operation.columnName)
|
|
.build().apply();
|
|
}
|
|
|
|
usedPartitionKeys.add(operation.partitionKey);
|
|
}
|
|
return usedPartitionKeys;
|
|
}
|
|
|
|
private static final long NEW_DELETION = FBUtilities.nowInSeconds();
|
|
private static final long PURGEABLE_DELETION = 42;
|
|
|
|
private enum OperationType
|
|
{
|
|
CREATE,
|
|
DELETE_PARTITION,
|
|
DELETE_RANGE,
|
|
DELETE_ROW,
|
|
DELETE_CELL
|
|
}
|
|
|
|
private static class TestWriteOperation
|
|
{
|
|
OperationType type;
|
|
String partitionKey;
|
|
String clusteringKey;
|
|
|
|
String clusteringRangeStart, clusteringRangeEnd;
|
|
String columnName;
|
|
String columnValue = "bla";
|
|
|
|
long deletionTime;
|
|
|
|
public TestWriteOperation(OperationType type, String partitionKey,
|
|
String clusteringKey, String clusteringRangeStart, String clusteringRangeEnd,
|
|
String columnName, long deletionTime)
|
|
{
|
|
this.type = type;
|
|
this.partitionKey = partitionKey;
|
|
this.clusteringKey = clusteringKey;
|
|
this.clusteringRangeStart = clusteringRangeStart;
|
|
this.clusteringRangeEnd = clusteringRangeEnd;
|
|
this.columnName = columnName;
|
|
this.deletionTime = deletionTime;
|
|
}
|
|
|
|
public static TestWriteOperation insert(String partitionKey, String clusteringKey,
|
|
String columnName)
|
|
{
|
|
return new TestWriteOperation(OperationType.CREATE, partitionKey, clusteringKey, null, null, columnName, 0);
|
|
}
|
|
|
|
public static TestWriteOperation deletePartition(String partitionKey, long deletionTime)
|
|
{
|
|
return new TestWriteOperation(OperationType.DELETE_PARTITION, partitionKey,
|
|
null, null, null, null, deletionTime);
|
|
}
|
|
|
|
public static TestWriteOperation deleteRange(String partitionKey, String clusteringRangeStart, String clusteringRangeEnd, long deletionTime)
|
|
{
|
|
return new TestWriteOperation(OperationType.DELETE_RANGE, partitionKey,
|
|
null, clusteringRangeStart, clusteringRangeEnd, null, deletionTime);
|
|
}
|
|
|
|
public static TestWriteOperation deleteRow(String partitionKey, String clusteringKey, long deletionTime)
|
|
{
|
|
return new TestWriteOperation(OperationType.DELETE_ROW, partitionKey, clusteringKey,
|
|
null, null, null, deletionTime);
|
|
}
|
|
|
|
public static TestWriteOperation deleteCell(String partitionKey, String clusteringKey, String columnName, long deletionTime)
|
|
{
|
|
return new TestWriteOperation(OperationType.DELETE_CELL, partitionKey, clusteringKey,
|
|
null, null, columnName, deletionTime);
|
|
}
|
|
|
|
|
|
}
|
|
|
|
private static void executeReadCommands(List<SinglePartitionReadCommand> commands) throws IOException
|
|
{
|
|
ReadQuery query = SinglePartitionReadCommand.Group.create(commands, DataLimits.NONE);
|
|
|
|
try (ReadExecutionController executionController = query.executionController();
|
|
UnfilteredPartitionIterator iter = query.executeLocally(executionController);
|
|
DataOutputBuffer buffer = new DataOutputBuffer())
|
|
{
|
|
UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter,
|
|
query.columnFilter(),
|
|
buffer,
|
|
MessagingService.current_version);
|
|
}
|
|
}
|
|
|
|
private static SinglePartitionReadCommand getWholePartitionReadCommand(ColumnFamilyStore cfs, String partitionKey, long nowInSeconds)
|
|
{
|
|
ColumnFilter columnFilter = ColumnFilter.allRegularColumnsBuilder(cfs.metadata(), false).build();
|
|
RowFilter rowFilter = RowFilter.create(true);
|
|
Slice slice = Slice.make(BufferClusteringBound.BOTTOM, BufferClusteringBound.TOP);
|
|
ClusteringIndexSliceFilter sliceFilter = new ClusteringIndexSliceFilter(Slices.with(cfs.metadata().comparator, slice), false);
|
|
return SinglePartitionReadCommand.create(cfs.metadata(), nowInSeconds,
|
|
columnFilter, rowFilter,
|
|
DataLimits.NONE, Util.dk(partitionKey), sliceFilter);
|
|
}
|
|
|
|
@Test
|
|
public void testSinglePartitionSliceRepairedDataTracking() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build();
|
|
testRepairedDataTracking(cfs, readCommand);
|
|
}
|
|
|
|
@Test
|
|
public void testPartitionRangeRepairedDataTracking() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
ReadCommand readCommand = Util.cmd(cfs).build();
|
|
testRepairedDataTracking(cfs, readCommand);
|
|
}
|
|
|
|
@Test
|
|
public void testSinglePartitionNamesRepairedDataTracking() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("cc").includeRow("dd").build();
|
|
testRepairedDataTracking(cfs, readCommand);
|
|
}
|
|
|
|
@Test
|
|
public void testSinglePartitionNamesSkipsOptimisationsIfTrackingRepairedData()
|
|
{
|
|
// when tracking, the optimizations of querying sstables in timestamp order and
|
|
// returning once all requested columns are not available so just assert that
|
|
// all sstables are read when performing such queries
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("dd")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 1, ByteBufferUtil.bytes("key"))
|
|
.clustering("dd")
|
|
.add("a", ByteBufferUtil.bytes("wxyz"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
List<SSTableReader> sstables = new ArrayList<>(cfs.getLiveSSTables());
|
|
assertEquals(2, sstables.size());
|
|
sstables.sort(SSTableReader.maxTimestampDescending);
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("dd").columns("a").build();
|
|
|
|
assertEquals(0, readCount(sstables.get(0)));
|
|
assertEquals(0, readCount(sstables.get(1)));
|
|
ReadCommand withTracking = readCommand.copy();
|
|
Util.getAll(withTracking, withTracking.executionController(true));
|
|
assertEquals(1, readCount(sstables.get(0)));
|
|
assertEquals(1, readCount(sstables.get(1)));
|
|
|
|
// same command without tracking touches only the table with the higher timestamp
|
|
Util.getAll(readCommand.copy());
|
|
assertEquals(2, readCount(sstables.get(0)));
|
|
assertEquals(1, readCount(sstables.get(1)));
|
|
}
|
|
|
|
@Test
|
|
public void dontIncludeLegacyCounterContextInDigest()
|
|
{
|
|
// Serializations of a CounterContext containing legacy (pre-2.1) shards
|
|
// can legitimately differ across replicas. For this reason, the context
|
|
// bytes are omitted from the repaired digest if they contain legacy shards.
|
|
// This clearly has a tradeoff with the efficacy of the digest, without doing
|
|
// so false positive digest mismatches will be reported for scenarios where
|
|
// there is nothing that can be done to "fix" the replicas
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF7);
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
|
|
// insert a row with the counter column having value 0, in a legacy shard.
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("aa")
|
|
.addLegacyCounterCell("c", 0L)
|
|
.build()
|
|
.apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
|
|
// execute a read and capture the digest
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build();
|
|
ByteBuffer digestWithLegacyCounter0 = performReadAndVerifyRepairedInfo(readCommand, 1, 1, true);
|
|
assertNotEquals(EMPTY_BYTE_BUFFER, digestWithLegacyCounter0);
|
|
|
|
// truncate, then re-insert the same partition, but this time with a legacy
|
|
// shard having the value 1. The repaired digest should match the previous, as
|
|
// the values (context) are not included, only the cell metadata (ttl, timestamp, etc)
|
|
cfs.truncateBlocking();
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("aa")
|
|
.addLegacyCounterCell("c", 1L)
|
|
.build()
|
|
.apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
|
|
ByteBuffer digestWithLegacyCounter1 = performReadAndVerifyRepairedInfo(readCommand, 1, 1, true);
|
|
assertEquals(digestWithLegacyCounter0, digestWithLegacyCounter1);
|
|
|
|
// truncate, then re-insert the same partition, but this time with a non-legacy
|
|
// counter cell present. The repaired digest should not match the previous ones
|
|
// as this time the value (context) is included.
|
|
cfs.truncateBlocking();
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("aa")
|
|
.add("c", 1L)
|
|
.build()
|
|
.apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
|
|
ByteBuffer digestWithCounterCell = performReadAndVerifyRepairedInfo(readCommand, 1, 1, true);
|
|
assertNotEquals(EMPTY_BYTE_BUFFER, digestWithCounterCell);
|
|
assertNotEquals(digestWithLegacyCounter0, digestWithCounterCell);
|
|
assertNotEquals(digestWithLegacyCounter1, digestWithCounterCell);
|
|
}
|
|
|
|
/**
|
|
* Writes a single partition containing a single row and reads using a partition read. The single
|
|
* row includes 1 live simple column, 1 simple tombstone and 1 complex column with a complex
|
|
* deletion and a live cell. The repaired data digests generated by executing the same query
|
|
* before and after the tombstones become eligible for purging should not match each other.
|
|
* Also, neither digest should be empty as the partition is not made empty by the purging.
|
|
*/
|
|
@Test
|
|
public void purgeGCableTombstonesBeforeCalculatingDigest()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF8);
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
setGCGrace(cfs, 600);
|
|
|
|
DecoratedKey[] keys = new DecoratedKey[] { Util.dk("key0"), Util.dk("key1"), Util.dk("key2"), Util.dk("key3") };
|
|
long nowInSec = FBUtilities.nowInSeconds();
|
|
|
|
// A simple tombstone
|
|
new RowUpdateBuilder(cfs.metadata(), 0, keys[0]).clustering("cc").delete("a").build().apply();
|
|
|
|
// Collection with an associated complex deletion
|
|
PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(cfs.metadata(), keys[1]).timestamp(0);
|
|
builder.row("cc").add("c", ImmutableSet.of("element1", "element2"));
|
|
builder.buildAsMutation().apply();
|
|
|
|
// RangeTombstone and a row (not covered by the RT). The row contains a regular tombstone which will not be
|
|
// purged. This is to prevent the partition from being fully purged and removed from the final results
|
|
new RowUpdateBuilder(cfs.metadata(), nowInSec, 0L, keys[2]).addRangeTombstone("aa", "bb").build().apply();
|
|
new RowUpdateBuilder(cfs.metadata(), nowInSec+ 1000, 1000L, keys[2]).clustering("cc").delete("a").build().apply();
|
|
|
|
// Partition with 2 rows, one fully deleted
|
|
new RowUpdateBuilder(cfs.metadata.get(), 0, keys[3]).clustering("bb").add("a", ByteBufferUtil.bytes("a")).delete("b").build().apply();
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, keys[3], "cc").apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
|
|
Map<DecoratedKey, ByteBuffer> digestsWithTombstones = new HashMap<>();
|
|
//Tombstones are not yet purgable
|
|
for (DecoratedKey key : keys)
|
|
{
|
|
ReadCommand cmd = Util.cmd(cfs, key).withNowInSeconds(nowInSec).build();
|
|
|
|
try (ReadExecutionController controller = cmd.executionController(true))
|
|
{
|
|
Partition partition = Util.getOnlyPartitionUnfiltered(cmd, controller);
|
|
assertFalse(partition.isEmpty());
|
|
partition.unfilteredIterator().forEachRemaining(u -> {
|
|
// must be either a RT, or a row containing some kind of deletion
|
|
assertTrue(u.isRangeTombstoneMarker() || ((Row) u).hasDeletion(cmd.nowInSec()));
|
|
});
|
|
ByteBuffer digestWithTombstones = controller.getRepairedDataDigest();
|
|
// None should generate an empty digest
|
|
assertDigestsDiffer(EMPTY_BYTE_BUFFER, digestWithTombstones);
|
|
digestsWithTombstones.put(key, digestWithTombstones);
|
|
}
|
|
}
|
|
|
|
// Make tombstones eligible for purging and re-run cmd with an incremented nowInSec
|
|
setGCGrace(cfs, 0);
|
|
|
|
//Tombstones are now purgable, so won't be in the read results and produce different digests
|
|
for (DecoratedKey key : keys)
|
|
{
|
|
ReadCommand cmd = Util.cmd(cfs, key).withNowInSeconds(nowInSec + 60).build();
|
|
|
|
try (ReadExecutionController controller = cmd.executionController(true))
|
|
{
|
|
Partition partition = Util.getOnlyPartitionUnfiltered(cmd, controller);
|
|
assertFalse(partition.isEmpty());
|
|
partition.unfilteredIterator().forEachRemaining(u -> {
|
|
// After purging, only rows without any deletions should remain.
|
|
// The one exception is "key2:cc" which has a regular column tombstone which is not
|
|
// eligible for purging. This is to prevent the partition from being fully purged
|
|
// when its RT is removed.
|
|
assertTrue(u.isRow());
|
|
Row r = (Row) u;
|
|
assertTrue(!r.hasDeletion(cmd.nowInSec())
|
|
|| (key.equals(keys[2]) && r.clustering()
|
|
.bufferAt(0)
|
|
.equals(AsciiType.instance.fromString("cc"))));
|
|
});
|
|
ByteBuffer digestWithoutTombstones = controller.getRepairedDataDigest();
|
|
// not an empty digest
|
|
assertDigestsDiffer(EMPTY_BYTE_BUFFER, digestWithoutTombstones);
|
|
// should not match the pre-purge digest
|
|
assertDigestsDiffer(digestsWithTombstones.get(key), digestWithoutTombstones);
|
|
}
|
|
}
|
|
}
|
|
|
|
@Test
|
|
public void testRepairedDataOverreadMetrics()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF9);
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
cfs.metadata().withSwapped(cfs.metadata().params.unbuild()
|
|
.caching(CachingParams.CACHE_NOTHING)
|
|
.build());
|
|
// Insert and repair
|
|
insert(cfs, IntStream.range(0, 10), () -> IntStream.range(0, 10));
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
// Insert and leave unrepaired
|
|
insert(cfs, IntStream.range(0, 10), () -> IntStream.range(10, 20));
|
|
|
|
// Single partition reads
|
|
int limit = 5;
|
|
ReadCommand cmd = Util.cmd(cfs, ByteBufferUtil.bytes(0)).withLimit(limit).build();
|
|
assertEquals(0, getAndResetOverreadCount(cfs));
|
|
|
|
// No overreads if not tracking
|
|
readAndCheckRowCount(Collections.singletonList(Util.getOnlyPartition(cmd)), limit);
|
|
assertEquals(0, getAndResetOverreadCount(cfs));
|
|
|
|
// Overread up to (limit - 1) if tracking is enabled
|
|
cmd = cmd.copy();
|
|
readAndCheckRowCount(Collections.singletonList(Util.getOnlyPartition(cmd, true)), limit);
|
|
// overread count is always < limit as the first read is counted during merging (and so is expected)
|
|
assertEquals(limit - 1, getAndResetOverreadCount(cfs));
|
|
|
|
// if limit already requires reading all repaired data, no overreads should be recorded
|
|
limit = 20;
|
|
cmd = Util.cmd(cfs, ByteBufferUtil.bytes(0)).withLimit(limit).build();
|
|
readAndCheckRowCount(Collections.singletonList(Util.getOnlyPartition(cmd)), limit);
|
|
assertEquals(0, getAndResetOverreadCount(cfs));
|
|
|
|
// Range reads
|
|
limit = 5;
|
|
cmd = Util.cmd(cfs).withLimit(limit).build();
|
|
assertEquals(0, getAndResetOverreadCount(cfs));
|
|
// No overreads if not tracking
|
|
readAndCheckRowCount(Util.getAll(cmd), limit);
|
|
assertEquals(0, getAndResetOverreadCount(cfs));
|
|
|
|
// Overread up to (limit - 1) if tracking is enabled
|
|
cmd = cmd.copy();
|
|
readAndCheckRowCount(Util.getAll(cmd, cmd.executionController(true)), limit);
|
|
assertEquals(limit - 1, getAndResetOverreadCount(cfs));
|
|
|
|
// if limit already requires reading all repaired data, no overreads should be recorded
|
|
limit = 100;
|
|
cmd = Util.cmd(cfs).withLimit(limit).build();
|
|
readAndCheckRowCount(Util.getAll(cmd), limit);
|
|
assertEquals(0, getAndResetOverreadCount(cfs));
|
|
}
|
|
|
|
private void setGCGrace(ColumnFamilyStore cfs, int gcGrace)
|
|
{
|
|
TableParams newParams = cfs.metadata().params.unbuild().gcGraceSeconds(gcGrace).build();
|
|
KeyspaceMetadata keyspaceMetadata = Schema.instance.getKeyspaceMetadata(cfs.metadata().keyspace);
|
|
SchemaTestUtil.addOrUpdateKeyspace(
|
|
keyspaceMetadata.withSwapped(
|
|
keyspaceMetadata.tables.withSwapped(
|
|
cfs.metadata().withSwapped(newParams))), true);
|
|
}
|
|
|
|
private long getAndResetOverreadCount(ColumnFamilyStore cfs)
|
|
{
|
|
// always clear the histogram after reading to make comparisons & asserts easier
|
|
long rows = cfs.metric.repairedDataTrackingOverreadRows.cf.getSnapshot().getMax();
|
|
((ClearableHistogram)cfs.metric.repairedDataTrackingOverreadRows.cf).clear();
|
|
return rows;
|
|
}
|
|
|
|
private void readAndCheckRowCount(Iterable<FilteredPartition> partitions, int expected)
|
|
{
|
|
int count = 0;
|
|
for (Partition partition : partitions)
|
|
{
|
|
assertFalse(partition.isEmpty());
|
|
try (UnfilteredRowIterator iter = partition.unfilteredIterator())
|
|
{
|
|
while (iter.hasNext())
|
|
{
|
|
iter.next();
|
|
count++;
|
|
}
|
|
}
|
|
}
|
|
assertEquals(expected, count);
|
|
}
|
|
|
|
private void insert(ColumnFamilyStore cfs, IntStream partitionIds, Supplier<IntStream> rowIds)
|
|
{
|
|
partitionIds.mapToObj(ByteBufferUtil::bytes)
|
|
.forEach( pk ->
|
|
rowIds.get().forEach( c ->
|
|
new RowUpdateBuilder(cfs.metadata(), 0, pk)
|
|
.clustering(c)
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply()
|
|
|
|
));
|
|
}
|
|
|
|
private void assertDigestsDiffer(ByteBuffer b0, ByteBuffer b1)
|
|
{
|
|
assertTrue(ByteBufferUtil.compareUnsigned(b0, b1) != 0);
|
|
}
|
|
|
|
@Test
|
|
public void partitionReadFullyPurged() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF6);
|
|
ReadCommand partitionRead = Util.cmd(cfs, Util.dk("key")).build();
|
|
fullyPurgedPartitionCreatesEmptyDigest(cfs, partitionRead);
|
|
}
|
|
|
|
@Test
|
|
public void rangeReadFullyPurged() throws Exception
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF6);
|
|
ReadCommand rangeRead = Util.cmd(cfs).build();
|
|
fullyPurgedPartitionCreatesEmptyDigest(cfs, rangeRead);
|
|
}
|
|
|
|
/**
|
|
* Writes a single partition containing only a single row deletion and reads with either a range or
|
|
* partition query. Before the row deletion is eligible for purging, it should appear in the query
|
|
* results and cause a non-empty repaired data digest to be generated. Repeating the query after
|
|
* the row deletion is eligible for purging, both the result set and the repaired data digest should
|
|
* be empty.
|
|
*/
|
|
private void fullyPurgedPartitionCreatesEmptyDigest(ColumnFamilyStore cfs, ReadCommand command)
|
|
{
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
setGCGrace(cfs, 600);
|
|
|
|
// Partition with a fully deleted static row and a single, fully deleted regular row
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, ByteBufferUtil.bytes("key")).apply();
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, ByteBufferUtil.bytes("key"), "cc").apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
|
|
try (ReadExecutionController controller = command.executionController(true))
|
|
{
|
|
List<ImmutableBTreePartition> partitions = Util.getAllUnfiltered(command, controller);
|
|
assertEquals(1, partitions.size());
|
|
ByteBuffer digestWithTombstones = controller.getRepairedDataDigest();
|
|
assertTrue(ByteBufferUtil.compareUnsigned(EMPTY_BYTE_BUFFER, digestWithTombstones) != 0);
|
|
|
|
// Make tombstones eligible for purging and re-run cmd with an incremented nowInSec
|
|
setGCGrace(cfs, 0);
|
|
}
|
|
|
|
AbstractReadCommandBuilder builder = command instanceof PartitionRangeReadCommand
|
|
? Util.cmd(cfs)
|
|
: Util.cmd(cfs, Util.dk("key"));
|
|
builder.withNowInSeconds(command.nowInSec() + 60);
|
|
command = builder.build();
|
|
|
|
try (ReadExecutionController controller = command.executionController(true))
|
|
{
|
|
List<ImmutableBTreePartition> partitions = Util.getAllUnfiltered(command, controller);
|
|
assertTrue(partitions.isEmpty());
|
|
ByteBuffer digestWithoutTombstones = controller.getRepairedDataDigest();
|
|
assertEquals(0, ByteBufferUtil.compareUnsigned(EMPTY_BYTE_BUFFER, digestWithoutTombstones));
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Verifies that during range reads which include multiple partitions, fully purged partitions
|
|
* have no material effect on the calculated digest. This test writes two sstables, each containing
|
|
* a single partition; the first is live and the second fully deleted and eligible for purging.
|
|
* Initially, only the sstable containing the live partition is marked repaired, while a range read
|
|
* which covers both partitions is performed to generate a digest. Then the sstable containing the
|
|
* purged partition is also marked repaired and the query reexecuted. The digests produced by both
|
|
* queries should match as the digest calculation should exclude the fully purged partition.
|
|
*/
|
|
@Test
|
|
public void mixedPurgedAndNonPurgedPartitions()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF6);
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
setGCGrace(cfs, 0);
|
|
|
|
ReadCommand command = Util.cmd(cfs).withNowInSeconds(FBUtilities.nowInSeconds() + 60).build();
|
|
|
|
// Live partition in a repaired sstable, so included in the digest calculation
|
|
new RowUpdateBuilder(cfs.metadata.get(), 0, ByteBufferUtil.bytes("key-0")).clustering("cc").add("a", ByteBufferUtil.bytes("a")).build().apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
// Fully deleted partition (static and regular rows) in an unrepaired sstable, so not included in the intial digest
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, ByteBufferUtil.bytes("key-1")).apply();
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, ByteBufferUtil.bytes("key-1"), "cc").apply();
|
|
Util.flush(cfs);
|
|
|
|
ByteBuffer digestWithoutPurgedPartition = null;
|
|
|
|
try (ReadExecutionController controller = command.executionController(true))
|
|
{
|
|
List<ImmutableBTreePartition> partitions = Util.getAllUnfiltered(command, controller);
|
|
assertEquals(1, partitions.size());
|
|
digestWithoutPurgedPartition = controller.getRepairedDataDigest();
|
|
assertTrue(ByteBufferUtil.compareUnsigned(EMPTY_BYTE_BUFFER, digestWithoutPurgedPartition) != 0);
|
|
}
|
|
|
|
// mark the sstable containing the purged partition as repaired, so both partitions are now
|
|
// read during in the digest calculation. Because the purged partition is entirely
|
|
// discarded, the resultant digest should match the earlier one.
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
command = Util.cmd(cfs).withNowInSeconds(command.nowInSec()).build();
|
|
|
|
try (ReadExecutionController controller = command.executionController(true))
|
|
{
|
|
List<ImmutableBTreePartition> partitions = Util.getAllUnfiltered(command, controller);
|
|
assertEquals(1, partitions.size());
|
|
ByteBuffer digestWithPurgedPartition = controller.getRepairedDataDigest();
|
|
assertEquals(0, ByteBufferUtil.compareUnsigned(digestWithPurgedPartition, digestWithoutPurgedPartition));
|
|
}
|
|
}
|
|
|
|
@Test
|
|
public void purgingConsidersRepairedDataOnly()
|
|
{
|
|
// 2 sstables, first is repaired and contains data that is all purgeable
|
|
// the second is unrepaired and contains non-purgable data. Even though
|
|
// the partition itself is not fully purged, the repaired data digest
|
|
// should be empty as there was no non-purgeable, repaired data read.
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF6);
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
setGCGrace(cfs, 0);
|
|
|
|
// Partition with a fully deleted static row and a single, fully deleted row which will be fully purged
|
|
DecoratedKey key = Util.dk("key");
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, key).apply();
|
|
RowUpdateBuilder.deleteRow(cfs.metadata(), 0, key, "cc").apply();
|
|
Util.flush(cfs);
|
|
cfs.getLiveSSTables().forEach(sstable -> mutateRepaired(cfs, sstable, 111, null));
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 1, key).clustering("cc").add("a", ByteBufferUtil.bytes("a")).build().apply();
|
|
Util.flush(cfs);
|
|
|
|
long nowInSec = FBUtilities.nowInSeconds() + 10;
|
|
ReadCommand cmd = Util.cmd(cfs, key).withNowInSeconds(nowInSec).build();
|
|
|
|
try (ReadExecutionController controller = cmd.executionController(true))
|
|
{
|
|
Partition partition = Util.getOnlyPartitionUnfiltered(cmd, controller);
|
|
assertFalse(partition.isEmpty());
|
|
// check that
|
|
try (UnfilteredRowIterator rows = partition.unfilteredIterator())
|
|
{
|
|
assertFalse(rows.isEmpty());
|
|
Unfiltered unfiltered = rows.next();
|
|
assertFalse(rows.hasNext());
|
|
assertTrue(unfiltered.isRow());
|
|
assertFalse(((Row) unfiltered).hasDeletion(nowInSec));
|
|
}
|
|
assertEquals(EMPTY_BYTE_BUFFER, controller.getRepairedDataDigest());
|
|
}
|
|
}
|
|
|
|
private long readCount(SSTableReader sstable)
|
|
{
|
|
return sstable.getReadMeter().count();
|
|
}
|
|
|
|
@Test
|
|
public void skipRowCacheIfTrackingRepairedData()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF6);
|
|
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("cc")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build();
|
|
assertTrue(cfs.isRowCacheEnabled());
|
|
// warm the cache
|
|
assertFalse(Util.getAll(readCommand).isEmpty());
|
|
long cacheHits = cfs.metric.rowCacheHit.getCount();
|
|
|
|
Util.getAll(readCommand);
|
|
assertTrue(cfs.metric.rowCacheHit.getCount() > cacheHits);
|
|
cacheHits = cfs.metric.rowCacheHit.getCount();
|
|
|
|
ReadCommand withRepairedInfo = readCommand.copy();
|
|
try (ReadExecutionController controller = withRepairedInfo.executionController(true))
|
|
{
|
|
Util.getAll(withRepairedInfo, controller);
|
|
assertEquals(cacheHits, cfs.metric.rowCacheHit.getCount());
|
|
}
|
|
}
|
|
|
|
@Test
|
|
public void testToCQLString()
|
|
{
|
|
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
|
|
DecoratedKey key = Util.dk("key");
|
|
|
|
ReadCommand readCommand = Util.cmd(cfs, key).build();
|
|
|
|
String result = readCommand.toCQLString();
|
|
|
|
assertEquals(result, String.format("SELECT * FROM \"ReadCommandTest\".\"Standard2\" WHERE key = 0x%s ALLOW FILTERING", ByteBufferUtil.bytesToHex(key.getKey())));
|
|
}
|
|
|
|
private void testRepairedDataTracking(ColumnFamilyStore cfs, ReadCommand readCommand)
|
|
{
|
|
cfs.truncateBlocking();
|
|
cfs.disableAutoCompaction();
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 0, ByteBufferUtil.bytes("key"))
|
|
.clustering("cc")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
|
|
new RowUpdateBuilder(cfs.metadata(), 1, ByteBufferUtil.bytes("key"))
|
|
.clustering("dd")
|
|
.add("a", ByteBufferUtil.bytes("abcd"))
|
|
.build()
|
|
.apply();
|
|
|
|
Util.flush(cfs);
|
|
List<SSTableReader> sstables = new ArrayList<>(cfs.getLiveSSTables());
|
|
assertEquals(2, sstables.size());
|
|
sstables.forEach(sstable -> assertFalse(sstable.isRepaired() || sstable.isPendingRepair()));
|
|
SSTableReader sstable1 = sstables.get(0);
|
|
SSTableReader sstable2 = sstables.get(1);
|
|
|
|
int numPartitions = 1;
|
|
int rowsPerPartition = 2;
|
|
|
|
// Capture all the digest versions as we mutate the table's repaired status. Each time
|
|
// we make a change, we expect a different digest.
|
|
Set<ByteBuffer> digests = new HashSet<>();
|
|
// first time round, nothing has been marked repaired so we expect digest to be an empty buffer and to be marked conclusive
|
|
ByteBuffer digest = performReadAndVerifyRepairedInfo(readCommand, numPartitions, rowsPerPartition, true);
|
|
assertEquals(EMPTY_BYTE_BUFFER, digest);
|
|
digests.add(digest);
|
|
|
|
// add a pending repair session to table1, digest should remain the same but now we expect it to be marked inconclusive
|
|
TimeUUID session1 = nextTimeUUID();
|
|
mutateRepaired(cfs, sstable1, ActiveRepairService.UNREPAIRED_SSTABLE, session1);
|
|
digests.add(performReadAndVerifyRepairedInfo(readCommand, numPartitions, rowsPerPartition, false));
|
|
assertEquals(1, digests.size());
|
|
|
|
// add a different pending session to table2, digest should remain the same and still consider it inconclusive
|
|
TimeUUID session2 = nextTimeUUID();
|
|
mutateRepaired(cfs, sstable2, ActiveRepairService.UNREPAIRED_SSTABLE, session2);
|
|
digests.add(performReadAndVerifyRepairedInfo(readCommand, numPartitions, rowsPerPartition, false));
|
|
assertEquals(1, digests.size());
|
|
|
|
// mark one table repaired
|
|
mutateRepaired(cfs, sstable1, 111, null);
|
|
// this time, digest should not be empty, session2 still means that the result is inconclusive
|
|
digests.add(performReadAndVerifyRepairedInfo(readCommand, numPartitions, rowsPerPartition, false));
|
|
assertEquals(2, digests.size());
|
|
|
|
// mark the second table repaired
|
|
mutateRepaired(cfs, sstable2, 222, null);
|
|
// digest should be updated again and as there are no longer any pending sessions, it should be considered conclusive
|
|
digests.add(performReadAndVerifyRepairedInfo(readCommand, numPartitions, rowsPerPartition, true));
|
|
assertEquals(3, digests.size());
|
|
|
|
// insert a partition tombstone into the memtable, then re-check the repaired info.
|
|
// This is to ensure that when the optimisations which skip reading from sstables
|
|
// when a newer partition tombstone has already been cause the digest to be marked
|
|
// as inconclusive.
|
|
// the exception to this case is for partition range reads, where we always read
|
|
// and generate digests for all sstables, so we only test this path for single partition reads
|
|
if (readCommand.isLimitedToOnePartition())
|
|
{
|
|
new Mutation(PartitionUpdate.simpleBuilder(cfs.metadata(), ByteBufferUtil.bytes("key"))
|
|
.delete()
|
|
.build()).apply();
|
|
digest = performReadAndVerifyRepairedInfo(readCommand, 0, rowsPerPartition, false);
|
|
assertEquals(EMPTY_BYTE_BUFFER, digest);
|
|
|
|
// now flush so we have an unrepaired table with the deletion and repeat the check
|
|
Util.flush(cfs);
|
|
digest = performReadAndVerifyRepairedInfo(readCommand, 0, rowsPerPartition, false);
|
|
assertEquals(EMPTY_BYTE_BUFFER, digest);
|
|
}
|
|
}
|
|
|
|
private void mutateRepaired(ColumnFamilyStore cfs, SSTableReader sstable, long repairedAt, TimeUUID pendingSession)
|
|
{
|
|
try
|
|
{
|
|
sstable.descriptor.getMetadataSerializer().mutateRepairMetadata(sstable.descriptor, repairedAt, pendingSession);
|
|
sstable.reloadSSTableMetadata();
|
|
}
|
|
catch (IOException e)
|
|
{
|
|
e.printStackTrace();
|
|
fail("Caught IOException when mutating sstable metadata");
|
|
}
|
|
|
|
if (pendingSession != null)
|
|
{
|
|
// setup a minimal repair session. This is necessary because we
|
|
// check for sessions which have exceeded timeout and been purged
|
|
Range<Token> range = new Range<>(cfs.metadata().partitioner.getMinimumToken(),
|
|
cfs.metadata().partitioner.getRandomToken());
|
|
ActiveRepairService.instance().registerParentRepairSession(pendingSession,
|
|
REPAIR_COORDINATOR,
|
|
Lists.newArrayList(cfs),
|
|
Sets.newHashSet(range),
|
|
true,
|
|
repairedAt,
|
|
true,
|
|
PreviewKind.NONE);
|
|
|
|
LocalSessionAccessor.prepareUnsafe(pendingSession, null, Sets.newHashSet(REPAIR_COORDINATOR));
|
|
}
|
|
}
|
|
|
|
private ByteBuffer performReadAndVerifyRepairedInfo(ReadCommand command,
|
|
int expectedPartitions,
|
|
int expectedRowsPerPartition,
|
|
boolean expectConclusive)
|
|
{
|
|
// perform equivalent read command multiple times and assert that
|
|
// the repaired data info is always consistent. Return the digest
|
|
// so we can verify that it changes when the repaired status of
|
|
// the queried tables does.
|
|
Set<ByteBuffer> digests = new HashSet<>();
|
|
for (int i = 0; i < 10; i++)
|
|
{
|
|
ReadCommand withRepairedInfo = command.copy();
|
|
|
|
try (ReadExecutionController controller = withRepairedInfo.executionController(true))
|
|
{
|
|
List<FilteredPartition> partitions = Util.getAll(withRepairedInfo, controller);
|
|
assertEquals(expectedPartitions, partitions.size());
|
|
partitions.forEach(p -> assertEquals(expectedRowsPerPartition, p.rowCount()));
|
|
|
|
ByteBuffer digest = controller.getRepairedDataDigest();
|
|
digests.add(digest);
|
|
assertEquals(1, digests.size());
|
|
assertEquals(expectConclusive, controller.isRepairedDataDigestConclusive());
|
|
}
|
|
}
|
|
return digests.iterator().next();
|
|
}
|
|
|
|
}
|