natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Ensure consistent view of partition columns between coordinator and replica in ColumnFilter 

Patch by Alex Petrov; reviewed by Aleksey Yeschenko for CASSANDRA-13004
This commit is contained in:
Alex Petrov 2017-05-31 17:01:14 +02:00
parent 7b9868ce27
commit 1f54aa424f
8 changed files with 192 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.14
* Ensure consistent view of partition columns between coordinator and replica in ColumnFilter (CASSANDRA-13004)
* Failed unregistering mbean during drop keyspace (CASSANDRA-13346)
* nodetool scrub/cleanup/upgradesstables exit code is wrong (CASSANDRA-13542)
* Fix the reported number of sstable data files accessed per read (CASSANDRA-13120)

23
NEWS.txt
View File

@ -18,6 +18,29 @@ using the provided 'sstableupgrade' tool.
Upgrading
---------
- ALTER TABLE (ADD/DROP COLUMN) operations concurrent with a read might
result into data corruption (see CASSANDRA-13004 for more details).
Fixing this bug required a messaging protocol version bump. By default,
Cassandra 3.0.14 will use 3014 version for messaging.
Since Schema Migrations rely the on exact messaging protocol version
match between nodes, if you need schema changes during the upgrade
process, you have to start your nodes with `-Dcassandra.force_3_0_protocol_version=true`
first, in order to temporarily force a backwards compatible protocol.
After the whole cluster is upgraded to 3.0.14, do a rolling
restart of the cluster without setting that flag.
3.0.14 nodes with and withouot the flag set will be able to do schema
migrations with other 3.x and 3.0.x releases.
While running the cluster with the flag set to true on 3.0.14 (in
compatibility mode), avoid adding or removing any columns to/from
existing tables.
If your cluster can do without schema migrations during the upgrade
time, just start the cluster normally without setting aforementioned
flag.
- If performing a rolling upgrade from 3.0.13, there will be a schema mismatch caused
by a bug with the schema digest calculation in 3.0.13. This will cause unnecessary
but otherwise harmless schema updates, see CASSANDRA-13559 for more details.

9
src/java/org/apache/cassandra/db/ReadResponse.java
View File

@ -378,7 +378,7 @@ public abstract class ReadResponse
if (digest.hasRemaining())
return new DigestResponse(digest);
assert version == MessagingService.VERSION_30;
assert version >= MessagingService.VERSION_30;
ByteBuffer data = ByteBufferUtil.readWithVIntLength(in);
return new RemoteDataResponse(data);
}
@ -413,9 +413,10 @@ public abstract class ReadResponse
long size = ByteBufferUtil.serializedSizeWithVIntLength(digest);
if (!isDigest)
{
// Note that we can only get there if version == 3.0, which is the current_version. When we'll change the
// version, we'll have to deserialize/re-serialize the data to be in the proper version.
assert version == MessagingService.VERSION_30;
// In theory, we should deserialize/re-serialize if the version asked is different from the current
// version as the content could have a different serialization format. So far though, we haven't made
// change to partition iterators serialization since 3.0 so we skip this.
assert version >= MessagingService.VERSION_30;
ByteBuffer data = ((DataResponse)response).data;
size += ByteBufferUtil.serializedSizeWithVIntLength(data);
}

2
src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
View File

@ -198,7 +198,7 @@ public class CommitLogDescriptor
case VERSION_22:
return MessagingService.VERSION_22;
case VERSION_30:
return MessagingService.VERSION_30;
return MessagingService.FORCE_3_0_PROTOCOL_VERSION ? MessagingService.VERSION_30 : MessagingService.VERSION_3014;
default:
throw new IllegalStateException("Unknown commitlog version " + version);
}

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

@ -31,6 +31,7 @@ import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
/**
* Represents which (non-PK) columns (and optionally which sub-part of a column for complex columns) are selected
@ -52,23 +53,27 @@ public class ColumnFilter
public static final Serializer serializer = new Serializer();
// Distinguish between the 2 cases described above: if 'isFetchAll' is true, then all columns will be retrieved
// by the query, but the values for column/cells not selected by 'selection' and 'subSelections' will be skipped.
// Otherwise, only the column/cells returned by 'selection' and 'subSelections' will be returned at all.
// by the query, but the values for column/cells not selected by 'queried' and 'subSelections' will be skipped.
// Otherwise, only the column/cells returned by 'queried' and 'subSelections' will be returned at all.
private final boolean isFetchAll;
private final CFMetaData metadata; // can be null if !isFetchAll
private final PartitionColumns selection; // can be null if isFetchAll and we don't want to skip any value
private final PartitionColumns queried; // can be null if isFetchAll and we don't want to skip any value
private final PartitionColumns fetched;
private final SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections; // can be null
/**
* Used on replica for deserialisation
*/
private ColumnFilter(boolean isFetchAll,
CFMetaData metadata,
PartitionColumns columns,
PartitionColumns fetched,
PartitionColumns queried,
SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections)
{
assert !isFetchAll || fetched != null;
assert isFetchAll || queried != null;
this.isFetchAll = isFetchAll;
this.metadata = metadata;
this.selection = columns;
this.fetched = isFetchAll ? fetched : queried;
this.queried = queried;
this.subSelections = subSelections;
}
@ -77,7 +82,7 @@ public class ColumnFilter
*/
public static ColumnFilter all(CFMetaData metadata)
{
return new ColumnFilter(true, metadata, null, null);
return new ColumnFilter(true, metadata.partitionColumns(), null, null);
}
/**
@ -98,7 +103,7 @@ public class ColumnFilter
*/
public static ColumnFilter selection(CFMetaData metadata, PartitionColumns queried)
{
return new ColumnFilter(true, metadata, queried, null);
return new ColumnFilter(true, metadata.partitionColumns(), queried, null);
}
/**
@ -111,7 +116,7 @@ public class ColumnFilter
*/
public PartitionColumns fetchedColumns()
{
return isFetchAll ? metadata.partitionColumns() : selection;
return fetched;
}
public boolean includesAllColumns()
@ -124,7 +129,7 @@ public class ColumnFilter
*/
public boolean includes(ColumnDefinition column)
{
return isFetchAll || selection.contains(column);
return isFetchAll || queried.contains(column);
}
/**
@ -301,7 +306,7 @@ public class ColumnFilter
boolean isFetchAll = metadata != null;
PartitionColumns selectedColumns = selection == null ? null : selection.build();
// It's only ok to have selection == null in ColumnFilter if isFetchAll. So deal with the case of a "selection" builder
// It's only ok to have queried == null in ColumnFilter if isFetchAll. So deal with the case of a "selection" builder
// with nothing selected (we can at least happen on some backward compatible queries - CASSANDRA-10471).
if (!isFetchAll && selectedColumns == null)
selectedColumns = PartitionColumns.NONE;
@ -314,20 +319,37 @@ public class ColumnFilter
s.put(subSelection.column().name, subSelection);
}
return new ColumnFilter(isFetchAll, metadata, selectedColumns, s);
return new ColumnFilter(isFetchAll, isFetchAll ? metadata.partitionColumns() : selectedColumns, selectedColumns, s);
}
}
@Override
public boolean equals(Object other)
{
if (other == this)
return true;
if (!(other instanceof ColumnFilter))
return false;
ColumnFilter otherCf = (ColumnFilter) other;
return otherCf.isFetchAll == this.isFetchAll &&
Objects.equals(otherCf.fetched, this.fetched) &&
Objects.equals(otherCf.queried, this.queried) &&
Objects.equals(otherCf.subSelections, this.subSelections);
}
@Override
public String toString()
{
if (isFetchAll)
return "*";
if (selection.isEmpty())
if (queried.isEmpty())
return "";
Iterator<ColumnDefinition> defs = selection.selectOrderIterator();
Iterator<ColumnDefinition> defs = queried.selectOrderIterator();
if (!defs.hasNext())
return "<none>";
@ -367,7 +389,7 @@ public class ColumnFilter
private static int makeHeaderByte(ColumnFilter selection)
{
return (selection.isFetchAll ? IS_FETCH_ALL_MASK : 0)
| (selection.selection != null ? HAS_SELECTION_MASK : 0)
| (selection.queried != null ? HAS_SELECTION_MASK : 0)
| (selection.subSelections != null ? HAS_SUB_SELECTIONS_MASK : 0);
}
@ -375,10 +397,16 @@ public class ColumnFilter
{
out.writeByte(makeHeaderByte(selection));
if (selection.selection != null)
if (version >= MessagingService.VERSION_3014 && selection.isFetchAll)
{
Columns.serializer.serialize(selection.selection.statics, out);
Columns.serializer.serialize(selection.selection.regulars, out);
Columns.serializer.serialize(selection.fetched.statics, out);
Columns.serializer.serialize(selection.fetched.regulars, out);
}
if (selection.queried != null)
{
Columns.serializer.serialize(selection.queried.statics, out);
Columns.serializer.serialize(selection.queried.regulars, out);
}
if (selection.subSelections != null)
@ -396,7 +424,23 @@ public class ColumnFilter
boolean hasSelection = (header & HAS_SELECTION_MASK) != 0;
boolean hasSubSelections = (header & HAS_SUB_SELECTIONS_MASK) != 0;
PartitionColumns fetched = null;
PartitionColumns selection = null;
if (isFetchAll)
{
if (version >= MessagingService.VERSION_3014)
{
Columns statics = Columns.serializer.deserialize(in, metadata);
Columns regulars = Columns.serializer.deserialize(in, metadata);
fetched = new PartitionColumns(statics, regulars);
}
else
{
fetched = metadata.partitionColumns();
}
}
if (hasSelection)
{
Columns statics = Columns.serializer.deserialize(in, metadata);
@ -416,17 +460,23 @@ public class ColumnFilter
}
}
return new ColumnFilter(isFetchAll, isFetchAll ? metadata : null, selection, subSelections);
return new ColumnFilter(isFetchAll, fetched, selection, subSelections);
}
public long serializedSize(ColumnFilter selection, int version)
{
long size = 1; // header byte
if (selection.selection != null)
if (version >= MessagingService.VERSION_3014 && selection.isFetchAll)
{
size += Columns.serializer.serializedSize(selection.selection.statics);
size += Columns.serializer.serializedSize(selection.selection.regulars);
size += Columns.serializer.serializedSize(selection.fetched.statics);
size += Columns.serializer.serializedSize(selection.fetched.regulars);
}
if (selection.queried != null)
{
size += Columns.serializer.serializedSize(selection.queried.statics);
size += Columns.serializer.serializedSize(selection.queried.regulars);
}
if (selection.subSelections != null)
@ -440,4 +490,4 @@ public class ColumnFilter
return size;
}
}
}
}

7
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -80,6 +80,10 @@ import org.apache.cassandra.utils.concurrent.SimpleCondition;
public final class MessagingService implements MessagingServiceMBean
{
// Required to allow schema migrations while upgrading within the minor 3.0.x versions to 3.0.14.
// See CASSANDRA-13004 for details.
public final static boolean FORCE_3_0_PROTOCOL_VERSION = Boolean.getBoolean("cassandra.force_3_0_protocol_version");
public static final String MBEAN_NAME = "org.apache.cassandra.net:type=MessagingService";
// 8 bits version, so don't waste versions
@ -88,7 +92,8 @@ public final class MessagingService implements MessagingServiceMBean
public static final int VERSION_21 = 8;
public static final int VERSION_22 = 9;
public static final int VERSION_30 = 10;
public static final int current_version = VERSION_30;
public static final int VERSION_3014 = 11;
public static final int current_version = FORCE_3_0_PROTOCOL_VERSION ? VERSION_30 : VERSION_3014;
public static final String FAILURE_CALLBACK_PARAM = "CAL_BAC";
public static final byte[] ONE_BYTE = new byte[1];

13
src/java/org/apache/cassandra/service/MigrationManager.java
View File

@ -144,10 +144,17 @@ public class MigrationManager
* Don't request schema from fat clients
*/
return MessagingService.instance().knowsVersion(endpoint)
&& MessagingService.instance().getRawVersion(endpoint) == MessagingService.current_version
&& is30Compatible(MessagingService.instance().getRawVersion(endpoint))
&& !Gossiper.instance.isGossipOnlyMember(endpoint);
}
// Since 3.0.14 protocol contains only a CASSANDRA-13004 bugfix, it is safe to accept schema changes
// from both 3.0 and 3.0.14.
private static boolean is30Compatible(int version)
{
return version == MessagingService.current_version || version == MessagingService.VERSION_3014;
}
public static boolean isReadyForBootstrap()
{
return MigrationTask.getInflightTasks().isEmpty();
@ -541,8 +548,8 @@ public class MigrationManager
{
// only push schema to nodes with known and equal versions
if (!endpoint.equals(FBUtilities.getBroadcastAddress()) &&
MessagingService.instance().knowsVersion(endpoint) &&
MessagingService.instance().getRawVersion(endpoint) == MessagingService.current_version)
MessagingService.instance().knowsVersion(endpoint) &&
is30Compatible(MessagingService.instance().getRawVersion(endpoint)))
pushSchemaMutation(endpoint, schema);
}

70
test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java Normal file
View File

@ -0,0 +1,70 @@
/*
* 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.filter;
import org.junit.Test;
import junit.framework.Assert;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.ByteBufferUtil;
public class ColumnFilterTest
{
final static ColumnFilter.Serializer serializer = new ColumnFilter.Serializer();
@Test
public void columnFilterSerialisationRoundTrip() throws Exception
{
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);
testRoundTrip(ColumnFilter.selection(metadata, metadata.partitionColumns().without(v1)), metadata, MessagingService.VERSION_30);
testRoundTrip(ColumnFilter.selection(metadata, metadata.partitionColumns().without(v1)), metadata, MessagingService.VERSION_3014);
}
static void testRoundTrip(ColumnFilter columnFilter, CFMetaData metadata, int version) throws Exception
{
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);
}
}