From 4f475d12e99f6760f139cf7fcb031a31152c2ff7 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 10:20:22 +0200 Subject: [PATCH 01/12] Validate arguments of blobAs functions patch by slebresne; reviewed by thobbs for CASSANDRA-7707 --- CHANGES.txt | 1 + .../cql3/functions/BytesConversionFcts.java | 22 ++++++++++++++---- .../cql3/functions/FunctionCall.java | 23 +++++++++++++++++-- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index df40933e01..b03e250676 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.0.10 + * Validate arguments of blobAs* functions (CASSANDRA-7707) * Fix potential AssertionError with 2ndary indexes (CASSANDRA-6612) * Avoid logging CompactionInterrupted at ERROR (CASSANDRA-7694) * Minor leak in sstable2jon (CASSANDRA-7709) diff --git a/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java b/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java index b30b5e72af..e3023db7bf 100644 --- a/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java +++ b/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java @@ -23,6 +23,9 @@ import java.util.List; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.serializers.MarshalException; public abstract class BytesConversionFcts { @@ -40,14 +43,25 @@ public abstract class BytesConversionFcts }; } - public static Function makeFromBlobFunction(AbstractType toType) + public static Function makeFromBlobFunction(final AbstractType toType) { - String name = "blobas" + toType.asCQL3Type(); + final String name = "blobas" + toType.asCQL3Type(); return new AbstractFunction(name, toType, BytesType.instance) { - public ByteBuffer execute(List parameters) + public ByteBuffer execute(List parameters) throws InvalidRequestException { - return parameters.get(0); + ByteBuffer val = parameters.get(0); + try + { + if (val != null) + toType.validate(val); + return val; + } + catch (MarshalException e) + { + throw new InvalidRequestException(String.format("In call to function %s, value 0x%s is not a valid binary representation for type %s", + name, ByteBufferUtil.bytesToHex(val), toType.asCQL3Type())); + } } }; } diff --git a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java b/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java index 3abf65e539..66e498f15f 100644 --- a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java +++ b/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java @@ -28,6 +28,8 @@ import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.db.marshal.MapType; import org.apache.cassandra.db.marshal.SetType; import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.serializers.MarshalException; public class FunctionCall extends Term.NonTerminal { @@ -63,8 +65,24 @@ public class FunctionCall extends Term.NonTerminal throw new InvalidRequestException(String.format("Invalid null value for argument to %s", fun)); buffers.add(val); } + return executeInternal(fun, buffers); + } - return fun.execute(buffers); + private static ByteBuffer executeInternal(Function fun, List params) throws InvalidRequestException + { + ByteBuffer result = fun.execute(params); + try + { + // Check the method didn't lied on it's declared return type + if (result != null) + fun.returnType().validate(result); + return result; + } + catch (MarshalException e) + { + throw new RuntimeException(String.format("Return of function %s (%s) is not a valid value for its declared return type %s", + fun, ByteBufferUtil.bytesToHex(result), fun.returnType().asCQL3Type())); + } } public boolean containsBindMarker() @@ -132,7 +150,8 @@ public class FunctionCall extends Term.NonTerminal assert t instanceof Term.Terminal; buffers.add(((Term.Terminal)t).get()); } - return fun.execute(buffers); + + return executeInternal(fun, buffers); } public boolean isAssignableTo(ColumnSpecification receiver) From 5bd37b92d2a8c75163ca51a0cfde908a299b7ca1 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 10:21:34 +0200 Subject: [PATCH 02/12] Fix potential AssertionError in RangeTombstoneList patch by slebresne; reviewed by thobbs for CASSANDRA-7700 --- CHANGES.txt | 1 + .../cassandra/db/RangeTombstoneList.java | 79 ++++++++++++++--- .../cassandra/db/RangeTombstoneListTest.java | 88 +++++++++++++++++++ 3 files changed, 157 insertions(+), 11 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index b03e250676..9c78d07356 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.0.10 + * Fix potential AssertionError in RangeTombstoneList (CASSANDRA-7700) * Validate arguments of blobAs* functions (CASSANDRA-7707) * Fix potential AssertionError with 2ndary indexes (CASSANDRA-6612) * Avoid logging CompactionInterrupted at ERROR (CASSANDRA-7694) diff --git a/src/java/org/apache/cassandra/db/RangeTombstoneList.java b/src/java/org/apache/cassandra/db/RangeTombstoneList.java index dad9004b54..dadcc20766 100644 --- a/src/java/org/apache/cassandra/db/RangeTombstoneList.java +++ b/src/java/org/apache/cassandra/db/RangeTombstoneList.java @@ -138,7 +138,7 @@ public class RangeTombstoneList implements Iterable { // Note: insertFrom expect i to be the insertion point in term of interval ends int pos = Arrays.binarySearch(ends, 0, size, start, comparator); - insertFrom((pos >= 0 ? pos+1 : -pos-1), start, end, markedAt, delTime); + insertFrom((pos >= 0 ? pos : -pos-1), start, end, markedAt, delTime); } } @@ -185,7 +185,7 @@ public class RangeTombstoneList implements Iterable int j = 0; while (i < size && j < tombstones.size) { - if (comparator.compare(tombstones.starts[j], ends[i]) < 0) + if (comparator.compare(tombstones.starts[j], ends[i]) <= 0) { insertFrom(i, tombstones.starts[j], tombstones.ends[j], tombstones.markedAts[j], tombstones.delTimes[j]); j++; @@ -380,16 +380,52 @@ public class RangeTombstoneList implements Iterable } /* - * Inserts a new element starting at index i. This method assumes that i is the insertion point - * in term of intervals for start: - * ends[i-1] <= start < ends[i] + * Inserts a new element starting at index i. This method assumes that: + * ends[i-1] <= start <= ends[i] + * + * A RangeTombstoneList is a list of range [s_0, e_0]...[s_n, e_n] such that: + * - s_i <= e_i + * - e_i <= s_i+1 + * - if s_i == e_i and e_i == s_i+1 then s_i+1 < e_i+1 + * Basically, range are non overlapping except for their bound and in order. And while + * we allow ranges with the same value for the start and end, we don't allow repeating + * such range (so we can't have [0, 0][0, 0] even though it would respect the first 2 + * conditions). + * */ private void insertFrom(int i, ByteBuffer start, ByteBuffer end, long markedAt, int delTime) { while (i < size) { - assert i == 0 || comparator.compare(start, ends[i-1]) >= 0; - assert i >= size || comparator.compare(start, ends[i]) < 0; + assert i == 0 || comparator.compare(ends[i-1], start) <= 0; + + int c = comparator.compare(start, ends[i]); + assert c <= 0; + if (c == 0) + { + // If start == ends[i], then we can insert from the next one (basically the new element + // really start at the next element), except for the case where starts[i] == ends[i]. + // In this latter case, if we were to move to next element, we could end up with ...[x, x][x, x]... + if (comparator.compare(starts[i], ends[i]) == 0) + { + // The current element cover a single value which is equal to the start of the inserted + // element. If the inserted element overwrites the current one, just remove the current + // (it's included in what we insert) and proceed with the insert. + if (markedAt > markedAts[i]) + { + removeInternal(i); + continue; + } + + // Otherwise (the current singleton interval override the new one), we want to leave the + // current element and move to the next, unless start == end since that means the new element + // is in fact fully covered by the current one (so we're done) + if (comparator.compare(start, end) == 0) + return; + } + i++; + continue; + } // Do we overwrite the current element? if (markedAt > markedAts[i]) @@ -407,11 +443,18 @@ public class RangeTombstoneList implements Iterable // now, start <= starts[i] - // If the new element stops before the current one, insert it and - // we're done - if (comparator.compare(end, starts[i]) <= 0) + // Does the new element stops before/at the current one, + int endCmp = comparator.compare(end, starts[i]); + if (endCmp <= 0) { - addInternal(i, start, end, markedAt, delTime); + // Here start <= starts[i] and end <= starts[i] + // This means the current element is before the current one. However, one special + // case is if end == starts[i] and starts[i] == ends[i]. In that case, + // the new element entirely overwrite the current one and we can just overwrite + if (endCmp == 0 && comparator.compare(starts[i], ends[i]) == 0) + setInternal(i, start, end, markedAt, delTime); + else + addInternal(i, start, end, markedAt, delTime); return; } @@ -503,6 +546,20 @@ public class RangeTombstoneList implements Iterable size++; } + private void removeInternal(int i) + { + assert i >= 0; + + System.arraycopy(starts, i+1, starts, i, size - i - 1); + System.arraycopy(ends, i+1, ends, i, size - i - 1); + System.arraycopy(markedAts, i+1, markedAts, i, size - i - 1); + System.arraycopy(delTimes, i+1, delTimes, i, size - i - 1); + + --size; + starts[size] = null; + ends[size] = null; + } + /* * Grow the arrays, leaving index i "free" in the process. */ diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java index dc9f9c4bf4..b0065e04d5 100644 --- a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java +++ b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java @@ -30,6 +30,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; public class RangeTombstoneListTest { private static final Comparator cmp = IntegerType.instance; + private static final Random rand = new Random(); @Test public void sortedAdditionTest() @@ -295,16 +296,103 @@ public class RangeTombstoneListTest assertEquals(6, l.maxMarkedAt()); } + private RangeTombstoneList makeRandom(int size, int maxItSize, int maxItDistance, int maxMarkedAt) + { + RangeTombstoneList l = new RangeTombstoneList(cmp, size); + + int prevStart = -1; + int prevEnd = 0; + for (int i = 0; i < size; i++) + { + int nextStart = prevEnd + rand.nextInt(maxItDistance); + int nextEnd = nextStart + rand.nextInt(maxItSize); + + // We can have an interval [x, x], but not 2 consecutives ones for the same x + if (nextEnd == nextStart && prevEnd == prevStart && prevEnd == nextStart) + nextEnd += 1 + rand.nextInt(maxItDistance); + + l.add(rt(nextStart, nextEnd, rand.nextInt(maxMarkedAt))); + + prevStart = nextStart; + prevEnd = nextEnd; + } + return l; + } + + @Test + public void addAllRandomTest() throws Throwable + { + int TEST_COUNT = 1000; + int MAX_LIST_SIZE = 50; + + int MAX_IT_SIZE = 20; + int MAX_IT_DISTANCE = 10; + int MAX_MARKEDAT = 10; + + for (int i = 0; i < TEST_COUNT; i++) + { + RangeTombstoneList l1 = makeRandom(rand.nextInt(MAX_LIST_SIZE) + 1, rand.nextInt(MAX_IT_SIZE) + 1, rand.nextInt(MAX_IT_DISTANCE) + 1, rand.nextInt(MAX_MARKEDAT) + 1); + RangeTombstoneList l2 = makeRandom(rand.nextInt(MAX_LIST_SIZE) + 1, rand.nextInt(MAX_IT_SIZE) + 1, rand.nextInt(MAX_IT_DISTANCE) + 1, rand.nextInt(MAX_MARKEDAT) + 1); + + RangeTombstoneList l1Initial = l1.copy(); + + try + { + // We generate the list randomly, so "all" we check is that the resulting range tombstone list looks valid. + l1.addAll(l2); + assertValid(l1); + } + catch (Throwable e) + { + System.out.println("Error merging:"); + System.out.println(" l1: " + toString(l1Initial)); + System.out.println(" l2: " + toString(l2)); + throw e; + } + } + } + private static void assertRT(RangeTombstone expected, RangeTombstone actual) { assertEquals(String.format("Expected %s but got %s", toString(expected), toString(actual)), expected, actual); } + private static void assertValid(RangeTombstoneList l) + { + // We check that ranges are in the right order and that we never have something + // like ...[x, x][x, x] ... + int prevStart = -2; + int prevEnd = -1; + for (RangeTombstone rt : l) + { + int curStart = i(rt.min); + int curEnd = i(rt.max); + + assertTrue("Invalid " + toString(l), prevEnd <= curStart); + assertTrue("Invalid " + toString(l), curStart <= curEnd); + + if (curStart == curEnd && prevEnd == curStart) + assertTrue("Invalid " + toString(l), prevStart != prevEnd); + + prevStart = curStart; + prevEnd = curEnd; + } + } + private static String toString(RangeTombstone rt) { return String.format("[%d, %d]@%d", i(rt.min), i(rt.max), rt.data.markedForDeleteAt); } + private static String toString(RangeTombstoneList l) + { + StringBuilder sb = new StringBuilder(); + sb.append("{"); + for (RangeTombstone rt : l) + sb.append(" ").append(toString(rt)); + return sb.append(" }").toString(); + } + private static ByteBuffer b(int i) { return ByteBufferUtil.bytes(i); From cd37d07baf5394d9bac6763de4556249e9837bb0 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 10:54:41 +0200 Subject: [PATCH 03/12] Version and licenses for 2.0.10 release --- .rat-excludes | 1 + NEWS.txt | 5 ++--- build.xml | 2 +- debian/changelog | 6 ++++++ .../cassandra/hadoop/pig/CqlNativeStorage.java | 17 +++++++++++++++++ ...QueryFilterWithStaticColumnsPresentTest.java | 17 +++++++++++++++++ 6 files changed, 44 insertions(+), 4 deletions(-) diff --git a/.rat-excludes b/.rat-excludes index 0da5ab9f97..503b3a6b92 100644 --- a/.rat-excludes +++ b/.rat-excludes @@ -30,3 +30,4 @@ examples/triggers/conf/* examples/hadoop_word_count/conf/log4j.properties pylib/cqlshlib/test/** src/resources/org/apache/cassandra/config/version.properties +**/hotspot_compiler diff --git a/NEWS.txt b/NEWS.txt index 04913842b7..7fa8be9c45 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -14,13 +14,12 @@ restore snapshots created with the previous major version using the using the provided 'sstableupgrade' tool. 2.0.10 -==== +====== New features ------------ - CqlPaginRecordReader and CqlPagingInputFormat have both been removed. Use CqlInputFormat instead. - - If you are using Leveled Compaction, you can now disable doing - size-tiered + - If you are using Leveled Compaction, you can now disable doing size-tiered compaction in L0 by starting Cassandra with -Dcassandra.disable_stcs_in_l0 (see CASSANDRA-6621 for details). - Shuffle and taketoken have been removed. For clusters that choose to diff --git a/build.xml b/build.xml index 4b64570e89..611345d80b 100644 --- a/build.xml +++ b/build.xml @@ -25,7 +25,7 @@ - + diff --git a/debian/changelog b/debian/changelog index 73fea4c974..e0b1eae726 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (2.0.10) unstable; urgency=medium + + * New release + + -- Sylvain Lebresne Fri, 08 Aug 2014 10:50:44 +0200 + cassandra (2.0.9) unstable; urgency=medium * New release diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java index 948d21c596..1e48bf4ead 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -1,3 +1,20 @@ +/* + * 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.hadoop.pig; import java.io.IOException; diff --git a/test/unit/org/apache/cassandra/cql3/SliceQueryFilterWithStaticColumnsPresentTest.java b/test/unit/org/apache/cassandra/cql3/SliceQueryFilterWithStaticColumnsPresentTest.java index a21ebdc2fa..75d1a1d5b6 100644 --- a/test/unit/org/apache/cassandra/cql3/SliceQueryFilterWithStaticColumnsPresentTest.java +++ b/test/unit/org/apache/cassandra/cql3/SliceQueryFilterWithStaticColumnsPresentTest.java @@ -1,3 +1,20 @@ +/* + * 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.cql3; import org.apache.cassandra.SchemaLoader; From 939994064bad12f6687ac1031b92e44a8f30bc11 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Thu, 7 Aug 2014 11:54:34 -0500 Subject: [PATCH 04/12] Fix pig tests. Patch by Alex Liu, reviewed by brandonwilliams for CASSANDRA-7570 --- .../cassandra/pig/CqlTableDataTypeTest.java | 73 ++++++++++-- .../apache/cassandra/pig/CqlTableTest.java | 106 +++++++++++++----- .../org/apache/cassandra/pig/PigTestBase.java | 3 + .../cassandra/pig/ThriftColumnFamilyTest.java | 41 +++++-- 4 files changed, 176 insertions(+), 47 deletions(-) diff --git a/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java b/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java index 2020b0a179..bbd5a873ef 100644 --- a/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java +++ b/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java @@ -217,10 +217,19 @@ public class CqlTableDataTypeTest extends PigTestBase } @Test - public void testCqlStorageRegularType() + public void testCqlNativeStorageRegularType() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { - pig.registerQuery("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from cqltable where token(key) > ? and token(key) <= ? + cqlTableTest("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20cqltable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + + //input_cql=select * from countertable where token(key) > ? and token(key) <= ? + counterTableTest("cc_rows = LOAD 'cql://cql3ks/countertable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20countertable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void cqlTableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); Iterator it = pig.openIterator("rows"); //{key: int, //col_ascii: chararray, @@ -257,21 +266,38 @@ public class CqlTableDataTypeTest extends PigTestBase Assert.assertEquals(t.get(14), "varchar"); Assert.assertEquals(t.get(15), 123); } - - pig.registerQuery("cc_rows = LOAD 'cql://cql3ks/countertable?" + defaultParameters + "' USING CqlStorage();"); - it = pig.openIterator("cc_rows"); + else + { + Assert.fail("Failed to get data for query " + initialQuery); + } + } + + private void counterTableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); + Iterator it = pig.openIterator("cc_rows"); if (it.hasNext()) { Tuple t = it.next(); Assert.assertEquals(t.get(0), 1); Assert.assertEquals(t.get(1), 3L); } + else + { + Assert.fail("Failed to get data for query " + initialQuery); + } } @Test - public void testCqlStorageSetType() + public void testCqlNativeStorageSetType() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { - pig.registerQuery("set_rows = LOAD 'cql://cql3ks/settable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from settable where token(key) > ? and token(key) <= ? + settableTest("set_rows = LOAD 'cql://cql3ks/settable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20settable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void settableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); Iterator it = pig.openIterator("set_rows"); if (it.hasNext()) { Tuple t = it.next(); @@ -322,13 +348,23 @@ public class CqlTableDataTypeTest extends PigTestBase Assert.assertEquals(innerTuple.get(0), 123); Assert.assertEquals(innerTuple.get(1), 124); } + else + { + Assert.fail("Failed to get data for query " + initialQuery); + } } @Test - public void testCqlStorageListType() + public void testCqlNativeStorageListType() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { - pig.registerQuery("list_rows = LOAD 'cql://cql3ks/listtable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from listtable where token(key) > ? and token(key) <= ? + listtableTest("list_rows = LOAD 'cql://cql3ks/listtable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20listtable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void listtableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); Iterator it = pig.openIterator("list_rows"); if (it.hasNext()) { Tuple t = it.next(); @@ -379,13 +415,23 @@ public class CqlTableDataTypeTest extends PigTestBase Assert.assertEquals(innerTuple.get(1), 123); Assert.assertEquals(innerTuple.get(0), 124); } + else + { + Assert.fail("Failed to get data for query " + initialQuery); + } } @Test - public void testCqlStorageMapType() + public void testCqlNativeStorageMapType() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { - pig.registerQuery("map_rows = LOAD 'cql://cql3ks/maptable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from maptable where token(key) > ? and token(key) <= ? + maptableTest("map_rows = LOAD 'cql://cql3ks/maptable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20maptable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void maptableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); Iterator it = pig.openIterator("map_rows"); if (it.hasNext()) { Tuple t = it.next(); @@ -436,5 +482,10 @@ public class CqlTableDataTypeTest extends PigTestBase Assert.assertEquals(innerTuple.get(0), 123); Assert.assertEquals(innerTuple.get(1), 124); } + else + { + Assert.fail("Failed to get data for query " + initialQuery); + } } + } diff --git a/test/pig/org/apache/cassandra/pig/CqlTableTest.java b/test/pig/org/apache/cassandra/pig/CqlTableTest.java index 15d49f27d9..4ca043da42 100644 --- a/test/pig/org/apache/cassandra/pig/CqlTableTest.java +++ b/test/pig/org/apache/cassandra/pig/CqlTableTest.java @@ -90,10 +90,36 @@ public class CqlTableTest extends PigTestBase } @Test - public void testCqlStorageSchema() + public void testCqlNativeStorageSchema() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { - pig.registerQuery("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from cqltable where token(key1) > ? and token(key1) <= ? + cqlTableSchemaTest("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20cqltable%20where%20token(key1)%20%3E%20%3F%20and%20token(key1)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + + //input_cql=select * from compactcqltable where token(key1) > ? and token(key1) <= ? + compactCqlTableSchemaTest("rows = LOAD 'cql://cql3ks/compactcqltable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compactcqltable%20where%20token(key1)%20%3E%20%3F%20and%20token(key1)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void compactCqlTableSchemaTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); + Iterator it = pig.openIterator("rows"); + if (it.hasNext()) { + Tuple t = it.next(); + Assert.assertEquals(t.get(0).toString(), "key1"); + Assert.assertEquals(t.get(1), 100); + Assert.assertEquals(t.get(2), 10.1f); + Assert.assertEquals(3, t.size()); + } + else + { + Assert.fail("Failed to get data for query " + initialQuery); + } + } + + private void cqlTableSchemaTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); Iterator it = pig.openIterator("rows"); if (it.hasNext()) { Tuple t = it.next(); @@ -103,26 +129,27 @@ public class CqlTableTest extends PigTestBase Assert.assertEquals(t.get(3), 10.1f); Assert.assertEquals(4, t.size()); } - - pig.registerQuery("rows = LOAD 'cql://cql3ks/compactcqltable?" + defaultParameters + "' USING CqlStorage();"); - it = pig.openIterator("rows"); - if (it.hasNext()) { - Tuple t = it.next(); - Assert.assertEquals(t.get(0).toString(), "key1"); - Assert.assertEquals(t.get(1), 100); - Assert.assertEquals(t.get(2), 10.1f); - Assert.assertEquals(3, t.size()); + else + { + Assert.fail("Failed to get data for query " + initialQuery); } } @Test - public void testCqlStorageSingleKeyTable() + public void testCqlNativeStorageSingleKeyTable() + throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException + { + //input_cql=select * from moredata where token(x) > ? and token(x) <= ? + SingleKeyTableTest("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20moredata%20where%20token(x)%20%3E%20%3F%20and%20token(x)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void SingleKeyTableTest(String initialQuery) throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { pig.setBatchOn(); - pig.registerQuery("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters + "' USING CqlStorage();"); + pig.registerQuery(initialQuery); pig.registerQuery("insertformat= FOREACH moretestvalues GENERATE TOTUPLE(TOTUPLE('a',x)),TOTUPLE(y);"); - pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/test?" + defaultParameters + "&output_query=UPDATE+cql3ks.test+set+b+%3D+%3F' USING CqlStorage();"); + pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/test?" + defaultParameters + nativeParameters + "&output_query=UPDATE+cql3ks.test+set+b+%3D+%3F' USING CqlNativeStorage();"); pig.executeBatch(); //(5,5) //(6,6) @@ -130,22 +157,30 @@ public class CqlTableTest extends PigTestBase //(2,2) //(3,3) //(1,1) - pig.registerQuery("result= LOAD 'cql://cql3ks/test?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from test where token(a) > ? and token(a) <= ? + pig.registerQuery("result= LOAD 'cql://cql3ks/test?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20test%20where%20token(a)%20%3E%20%3F%20and%20token(a)%20%3C%3D%20%3F' USING CqlNativeStorage();"); Iterator it = pig.openIterator("result"); - if (it.hasNext()) { + while (it.hasNext()) { Tuple t = it.next(); Assert.assertEquals(t.get(0), t.get(1)); } } @Test - public void testCqlStorageCompositeKeyTable() + public void testCqlNativeStorageCompositeKeyTable() + throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException + { + //input_cql=select * from compmore where token(id) > ? and token(id) <= ? + CompositeKeyTableTest("moredata= LOAD 'cql://cql3ks/compmore?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compmore%20where%20token(id)%20%3E%20%3F%20and%20token(id)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void CompositeKeyTableTest(String initialQuery) throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { pig.setBatchOn(); - pig.registerQuery("moredata= LOAD 'cql://cql3ks/compmore?" + defaultParameters + "' USING CqlStorage();"); + pig.registerQuery(initialQuery); pig.registerQuery("insertformat = FOREACH moredata GENERATE TOTUPLE (TOTUPLE('a',x),TOTUPLE('b',y), TOTUPLE('c',z)),TOTUPLE(data);"); - pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/compotable?" + defaultParameters + "&output_query=UPDATE%20cql3ks.compotable%20SET%20d%20%3D%20%3F' USING CqlStorage();"); + pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/compotable?" + defaultParameters + nativeParameters + "&output_query=UPDATE%20cql3ks.compotable%20SET%20d%20%3D%20%3F' USING CqlNativeStorage();"); pig.executeBatch(); //(5,6,Fix,nomatch) @@ -157,7 +192,8 @@ public class CqlTableTest extends PigTestBase //(6,5,Sive,nomatch) //(4,4,Four,match) //(9,10,Ninen,nomatch) - pig.registerQuery("result= LOAD 'cql://cql3ks/compotable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from compotable where token(a) > ? and token(a) <= ? + pig.registerQuery("result= LOAD 'cql://cql3ks/compotable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compotable%20where%20token(a)%20%3E%20%3F%20and%20token(a)%20%3C%3D%20%3F' USING CqlNativeStorage();"); Iterator it = pig.openIterator("result"); int count = 0; while (it.hasNext()) { @@ -168,22 +204,30 @@ public class CqlTableTest extends PigTestBase } @Test - public void testCqlStorageCollectionColumnTable() + public void testCqlNativeStorageCollectionColumnTable() + throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException + { + //input_cql=select * from collectiontable where token(m) > ? and token(m) <= ? + CollectionColumnTableTest("collectiontable= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20collectiontable%20where%20token(m)%20%3E%20%3F%20and%20token(m)%20%3C%3D%20%3F' USING CqlNativeStorage();"); + } + + private void CollectionColumnTableTest(String initialQuery) throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { pig.setBatchOn(); - pig.registerQuery("collectiontable= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters + "' USING CqlStorage();"); + pig.registerQuery(initialQuery); pig.registerQuery("recs= FOREACH collectiontable GENERATE TOTUPLE(TOTUPLE('m', m) ), TOTUPLE(TOTUPLE('map', TOTUPLE('m', 'mm'), TOTUPLE('n', 'nn')));"); - pig.registerQuery("STORE recs INTO 'cql://cql3ks/collectiontable?" + defaultParameters + "&output_query=update+cql3ks.collectiontable+set+n+%3D+%3F' USING CqlStorage();"); + pig.registerQuery("STORE recs INTO 'cql://cql3ks/collectiontable?" + defaultParameters + nativeParameters + "&output_query=update+cql3ks.collectiontable+set+n+%3D+%3F' USING CqlNativeStorage();"); pig.executeBatch(); //(book2,((m,mm),(n,nn))) //(book3,((m,mm),(n,nn))) //(book4,((m,mm),(n,nn))) //(book1,((m,mm),(n,nn))) - pig.registerQuery("result= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters + "' USING CqlStorage();"); + //input_cql=select * from collectiontable where token(m) > ? and token(m) <= ? + pig.registerQuery("result= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20collectiontable%20where%20token(m)%20%3E%20%3F%20and%20token(m)%20%3C%3D%20%3F' USING CqlNativeStorage();"); Iterator it = pig.openIterator("result"); - while (it.hasNext()) { + if (it.hasNext()) { Tuple t = it.next(); Tuple t1 = (Tuple) t.get(1); Assert.assertEquals(t1.size(), 2); @@ -194,6 +238,10 @@ public class CqlTableTest extends PigTestBase Assert.assertEquals(element2.get(0), "n"); Assert.assertEquals(element2.get(1), "nn"); } + else + { + Assert.fail("Can't fetch any data"); + } } @Test @@ -236,6 +284,10 @@ public class CqlTableTest extends PigTestBase } Assert.assertEquals(3, columns.size()); } + else + { + Assert.fail("Can't fetch any data"); + } //results: (key1,(column1,100),(column2,10.1)) pig.registerQuery("compact_rows = LOAD 'cassandra://cql3ks/compactcqltable?" + defaultParameters + "' USING CassandraStorage();"); @@ -253,5 +305,9 @@ public class CqlTableTest extends PigTestBase Assert.assertEquals(column.get(0), "column2"); Assert.assertEquals(column.get(1), 10.1f); } + else + { + Assert.fail("Can't fetch any data"); + } } } diff --git a/test/pig/org/apache/cassandra/pig/PigTestBase.java b/test/pig/org/apache/cassandra/pig/PigTestBase.java index 83dc63b46c..002fbbab23 100644 --- a/test/pig/org/apache/cassandra/pig/PigTestBase.java +++ b/test/pig/org/apache/cassandra/pig/PigTestBase.java @@ -66,6 +66,9 @@ public class PigTestBase extends SchemaLoader protected static MiniCluster cluster; protected static PigServer pig; protected static String defaultParameters= "init_address=localhost&rpc_port=9170&partitioner=org.apache.cassandra.dht.ByteOrderedPartitioner"; + protected static String nativeParameters = "&core_conns=2&max_conns=10&min_simult_reqs=3&max_simult_reqs=10&native_timeout=10000000" + + "&native_read_timeout=10000000&send_buff_size=4096&receive_buff_size=4096&solinger=3" + + "&tcp_nodelay=true&reuse_address=true&keep_alive=true&native_port=9052"; static { diff --git a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java index 60344d2844..8903297128 100644 --- a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java +++ b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java @@ -200,10 +200,24 @@ public class ThriftColumnFamilyTest extends PigTestBase } @Test - public void testCqlStorage() throws IOException, ClassNotFoundException, TException, TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException, IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException + public void testCqlNativeStorage() throws IOException, ClassNotFoundException, TException, TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException, IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException { //regular thrift column families - pig.registerQuery("data = load 'cql://thriftKs/SomeApp?" + defaultParameters + "' using CqlStorage();"); + //input_cql=select * from "SomeApp" where token(key) > ? and token(key) <= ? + cqlStorageTest("data = load 'cql://thriftKs/SomeApp?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20%22SomeApp%22%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();"); + + //Test counter colun family + //input_cql=select * from "CC" where token(key) > ? and token(key) <= ? + cqlStorageCounterTableTest("cc_data = load 'cql://thriftKs/CC?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20%22CC%22%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();"); + + //Test composite column family + //input_cql=select * from "Compo" where token(key) > ? and token(key) <= ? + cqlStorageCompositeTableTest("compo_data = load 'cql://thriftKs/Compo?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20%22Compo%22%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F' using CqlNativeStorage();"); + } + + private void cqlStorageTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); //(bar,3.141592653589793,1335890877,User Bar,35.0,9,15000,like) //(baz,1.61803399,1335890877,User Baz,95.3,3,512000,dislike) @@ -256,16 +270,18 @@ public class ThriftColumnFamilyTest extends PigTestBase } } Assert.assertEquals(count, 4); + } - //Test counter colun family - pig.registerQuery("cc_data = load 'cql://thriftKs/CC?" + defaultParameters + "' using CqlStorage();"); + private void cqlStorageCounterTableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); //(chuck,fist,1) //(chuck,kick,3) // {key: chararray,column1: chararray,value: long} - it = pig.openIterator("cc_data"); - count = 0; + Iterator it = pig.openIterator("cc_data"); + int count = 0; while (it.hasNext()) { count ++; Tuple t = it.next(); @@ -275,9 +291,11 @@ public class ThriftColumnFamilyTest extends PigTestBase Assert.assertEquals(t.get(2), 3L); } Assert.assertEquals(count, 2); + } - //Test composite column family - pig.registerQuery("compo_data = load 'cql://thriftKs/Compo?" + defaultParameters + "' using CqlStorage();"); + private void cqlStorageCompositeTableTest(String initialQuery) throws IOException + { + pig.registerQuery(initialQuery); //(kick,bruce,bruce,watch it, mate) //(kick,bruce,lee,oww) @@ -285,8 +303,8 @@ public class ThriftColumnFamilyTest extends PigTestBase //(punch,bruce,lee,ouch) //{key: chararray,column1: chararray,column2: chararray,value: chararray} - it = pig.openIterator("compo_data"); - count = 0; + Iterator it = pig.openIterator("compo_data"); + int count = 0; while (it.hasNext()) { count ++; Tuple t = it.next(); @@ -583,7 +601,8 @@ public class ThriftColumnFamilyTest extends PigTestBase } } - @Test + /** This test case fails due to antlr lib conflicts, Cassandra2.1 uses 3.2, Hive1.2 uses 3.4 */ + //@Test public void testCassandraStorageCompositeColumnCF() throws IOException, ClassNotFoundException, TException, TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException, IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException { //Test CompositeType From b70eaf380a4560e2d241125c8c71fa23fee4877f Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 10:20:22 +0200 Subject: [PATCH 05/12] Validate arguments of blobAs functions patch by slebresne; reviewed by thobbs for CASSANDRA-7707 --- CHANGES.txt | 1 + .../cql3/functions/BytesConversionFcts.java | 22 ++++++++++++++---- .../cql3/functions/FunctionCall.java | 23 +++++++++++++++++-- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 9ebb8cd66e..2c2ab71414 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -6,6 +6,7 @@ * Fix UDT field selection with empty fields (CASSANDRA-7670) * Bogus deserialization of static cells from sstable (CASSANDRA-7684) Merged from 2.0: + * Validate arguments of blobAs* functions (CASSANDRA-7707) * Fix potential AssertionError with 2ndary indexes (CASSANDRA-6612) * Avoid logging CompactionInterrupted at ERROR (CASSANDRA-7694) * Minor leak in sstable2jon (CASSANDRA-7709) diff --git a/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java b/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java index b30b5e72af..e3023db7bf 100644 --- a/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java +++ b/src/java/org/apache/cassandra/cql3/functions/BytesConversionFcts.java @@ -23,6 +23,9 @@ import java.util.List; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.serializers.MarshalException; public abstract class BytesConversionFcts { @@ -40,14 +43,25 @@ public abstract class BytesConversionFcts }; } - public static Function makeFromBlobFunction(AbstractType toType) + public static Function makeFromBlobFunction(final AbstractType toType) { - String name = "blobas" + toType.asCQL3Type(); + final String name = "blobas" + toType.asCQL3Type(); return new AbstractFunction(name, toType, BytesType.instance) { - public ByteBuffer execute(List parameters) + public ByteBuffer execute(List parameters) throws InvalidRequestException { - return parameters.get(0); + ByteBuffer val = parameters.get(0); + try + { + if (val != null) + toType.validate(val); + return val; + } + catch (MarshalException e) + { + throw new InvalidRequestException(String.format("In call to function %s, value 0x%s is not a valid binary representation for type %s", + name, ByteBufferUtil.bytesToHex(val), toType.asCQL3Type())); + } } }; } diff --git a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java b/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java index a0c7447bcb..4ae7c987fa 100644 --- a/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java +++ b/src/java/org/apache/cassandra/cql3/functions/FunctionCall.java @@ -28,6 +28,8 @@ import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.db.marshal.MapType; import org.apache.cassandra.db.marshal.SetType; import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.serializers.MarshalException; public class FunctionCall extends Term.NonTerminal { @@ -63,8 +65,24 @@ public class FunctionCall extends Term.NonTerminal throw new InvalidRequestException(String.format("Invalid null value for argument to %s", fun)); buffers.add(val); } + return executeInternal(fun, buffers); + } - return fun.execute(buffers); + private static ByteBuffer executeInternal(Function fun, List params) throws InvalidRequestException + { + ByteBuffer result = fun.execute(params); + try + { + // Check the method didn't lied on it's declared return type + if (result != null) + fun.returnType().validate(result); + return result; + } + catch (MarshalException e) + { + throw new RuntimeException(String.format("Return of function %s (%s) is not a valid value for its declared return type %s", + fun, ByteBufferUtil.bytesToHex(result), fun.returnType().asCQL3Type())); + } } public boolean containsBindMarker() @@ -132,7 +150,8 @@ public class FunctionCall extends Term.NonTerminal assert t instanceof Term.Terminal; buffers.add(((Term.Terminal)t).get(QueryOptions.DEFAULT)); } - return fun.execute(buffers); + + return executeInternal(fun, buffers); } public boolean isAssignableTo(String keyspace, ColumnSpecification receiver) From 78952731f320ed792783deea988164389794bb30 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 10:21:34 +0200 Subject: [PATCH 06/12] Fix potential AssertionError in RangeTombstoneList patch by slebresne; reviewed by thobbs for CASSANDRA-7700 --- CHANGES.txt | 1 + .../cassandra/db/RangeTombstoneList.java | 79 ++++++++++++++--- .../cassandra/db/RangeTombstoneListTest.java | 88 +++++++++++++++++++ 3 files changed, 157 insertions(+), 11 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 2c2ab71414..fff06ae536 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -6,6 +6,7 @@ * Fix UDT field selection with empty fields (CASSANDRA-7670) * Bogus deserialization of static cells from sstable (CASSANDRA-7684) Merged from 2.0: + * Fix potential AssertionError in RangeTombstoneList (CASSANDRA-7700) * Validate arguments of blobAs* functions (CASSANDRA-7707) * Fix potential AssertionError with 2ndary indexes (CASSANDRA-6612) * Avoid logging CompactionInterrupted at ERROR (CASSANDRA-7694) diff --git a/src/java/org/apache/cassandra/db/RangeTombstoneList.java b/src/java/org/apache/cassandra/db/RangeTombstoneList.java index 393ff3c101..cbb4abe931 100644 --- a/src/java/org/apache/cassandra/db/RangeTombstoneList.java +++ b/src/java/org/apache/cassandra/db/RangeTombstoneList.java @@ -168,7 +168,7 @@ public class RangeTombstoneList implements Iterable, IMeasurable { // Note: insertFrom expect i to be the insertion point in term of interval ends int pos = Arrays.binarySearch(ends, 0, size, start, comparator); - insertFrom((pos >= 0 ? pos+1 : -pos-1), start, end, markedAt, delTime); + insertFrom((pos >= 0 ? pos : -pos-1), start, end, markedAt, delTime); } boundaryHeapSize += start.unsharedHeapSize() + end.unsharedHeapSize(); } @@ -216,7 +216,7 @@ public class RangeTombstoneList implements Iterable, IMeasurable int j = 0; while (i < size && j < tombstones.size) { - if (comparator.compare(tombstones.starts[j], ends[i]) < 0) + if (comparator.compare(tombstones.starts[j], ends[i]) <= 0) { insertFrom(i, tombstones.starts[j], tombstones.ends[j], tombstones.markedAts[j], tombstones.delTimes[j]); j++; @@ -517,16 +517,52 @@ public class RangeTombstoneList implements Iterable, IMeasurable } /* - * Inserts a new element starting at index i. This method assumes that i is the insertion point - * in term of intervals for start: - * ends[i-1] <= start < ends[i] + * Inserts a new element starting at index i. This method assumes that: + * ends[i-1] <= start <= ends[i] + * + * A RangeTombstoneList is a list of range [s_0, e_0]...[s_n, e_n] such that: + * - s_i <= e_i + * - e_i <= s_i+1 + * - if s_i == e_i and e_i == s_i+1 then s_i+1 < e_i+1 + * Basically, range are non overlapping except for their bound and in order. And while + * we allow ranges with the same value for the start and end, we don't allow repeating + * such range (so we can't have [0, 0][0, 0] even though it would respect the first 2 + * conditions). + * */ private void insertFrom(int i, Composite start, Composite end, long markedAt, int delTime) { while (i < size) { - assert i == 0 || comparator.compare(start, ends[i-1]) >= 0; - assert i >= size || comparator.compare(start, ends[i]) < 0; + assert i == 0 || comparator.compare(ends[i-1], start) <= 0; + + int c = comparator.compare(start, ends[i]); + assert c <= 0; + if (c == 0) + { + // If start == ends[i], then we can insert from the next one (basically the new element + // really start at the next element), except for the case where starts[i] == ends[i]. + // In this latter case, if we were to move to next element, we could end up with ...[x, x][x, x]... + if (comparator.compare(starts[i], ends[i]) == 0) + { + // The current element cover a single value which is equal to the start of the inserted + // element. If the inserted element overwrites the current one, just remove the current + // (it's included in what we insert) and proceed with the insert. + if (markedAt > markedAts[i]) + { + removeInternal(i); + continue; + } + + // Otherwise (the current singleton interval override the new one), we want to leave the + // current element and move to the next, unless start == end since that means the new element + // is in fact fully covered by the current one (so we're done) + if (comparator.compare(start, end) == 0) + return; + } + i++; + continue; + } // Do we overwrite the current element? if (markedAt > markedAts[i]) @@ -544,11 +580,18 @@ public class RangeTombstoneList implements Iterable, IMeasurable // now, start <= starts[i] - // If the new element stops before the current one, insert it and - // we're done - if (comparator.compare(end, starts[i]) <= 0) + // Does the new element stops before/at the current one, + int endCmp = comparator.compare(end, starts[i]); + if (endCmp <= 0) { - addInternal(i, start, end, markedAt, delTime); + // Here start <= starts[i] and end <= starts[i] + // This means the current element is before the current one. However, one special + // case is if end == starts[i] and starts[i] == ends[i]. In that case, + // the new element entirely overwrite the current one and we can just overwrite + if (endCmp == 0 && comparator.compare(starts[i], ends[i]) == 0) + setInternal(i, start, end, markedAt, delTime); + else + addInternal(i, start, end, markedAt, delTime); return; } @@ -640,6 +683,20 @@ public class RangeTombstoneList implements Iterable, IMeasurable size++; } + private void removeInternal(int i) + { + assert i >= 0; + + System.arraycopy(starts, i+1, starts, i, size - i - 1); + System.arraycopy(ends, i+1, ends, i, size - i - 1); + System.arraycopy(markedAts, i+1, markedAts, i, size - i - 1); + System.arraycopy(delTimes, i+1, delTimes, i, size - i - 1); + + --size; + starts[size] = null; + ends[size] = null; + } + /* * Grow the arrays, leaving index i "free" in the process. */ diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java index faa15f0ee1..798ce91c9f 100644 --- a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java +++ b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java @@ -31,6 +31,7 @@ import org.apache.cassandra.utils.ByteBufferUtil; public class RangeTombstoneListTest { private static final Comparator cmp = new SimpleDenseCellNameType(IntegerType.instance); + private static final Random rand = new Random(); @Test public void sortedAdditionTest() @@ -290,16 +291,103 @@ public class RangeTombstoneListTest assertEquals(6, l.maxMarkedAt()); } + private RangeTombstoneList makeRandom(int size, int maxItSize, int maxItDistance, int maxMarkedAt) + { + RangeTombstoneList l = new RangeTombstoneList(cmp, size); + + int prevStart = -1; + int prevEnd = 0; + for (int i = 0; i < size; i++) + { + int nextStart = prevEnd + rand.nextInt(maxItDistance); + int nextEnd = nextStart + rand.nextInt(maxItSize); + + // We can have an interval [x, x], but not 2 consecutives ones for the same x + if (nextEnd == nextStart && prevEnd == prevStart && prevEnd == nextStart) + nextEnd += 1 + rand.nextInt(maxItDistance); + + l.add(rt(nextStart, nextEnd, rand.nextInt(maxMarkedAt))); + + prevStart = nextStart; + prevEnd = nextEnd; + } + return l; + } + + @Test + public void addAllRandomTest() throws Throwable + { + int TEST_COUNT = 1000; + int MAX_LIST_SIZE = 50; + + int MAX_IT_SIZE = 20; + int MAX_IT_DISTANCE = 10; + int MAX_MARKEDAT = 10; + + for (int i = 0; i < TEST_COUNT; i++) + { + RangeTombstoneList l1 = makeRandom(rand.nextInt(MAX_LIST_SIZE) + 1, rand.nextInt(MAX_IT_SIZE) + 1, rand.nextInt(MAX_IT_DISTANCE) + 1, rand.nextInt(MAX_MARKEDAT) + 1); + RangeTombstoneList l2 = makeRandom(rand.nextInt(MAX_LIST_SIZE) + 1, rand.nextInt(MAX_IT_SIZE) + 1, rand.nextInt(MAX_IT_DISTANCE) + 1, rand.nextInt(MAX_MARKEDAT) + 1); + + RangeTombstoneList l1Initial = l1.copy(); + + try + { + // We generate the list randomly, so "all" we check is that the resulting range tombstone list looks valid. + l1.addAll(l2); + assertValid(l1); + } + catch (Throwable e) + { + System.out.println("Error merging:"); + System.out.println(" l1: " + toString(l1Initial)); + System.out.println(" l2: " + toString(l2)); + throw e; + } + } + } + private static void assertRT(RangeTombstone expected, RangeTombstone actual) { assertEquals(String.format("Expected %s but got %s", toString(expected), toString(actual)), expected, actual); } + private static void assertValid(RangeTombstoneList l) + { + // We check that ranges are in the right order and that we never have something + // like ...[x, x][x, x] ... + int prevStart = -2; + int prevEnd = -1; + for (RangeTombstone rt : l) + { + int curStart = i(rt.min); + int curEnd = i(rt.max); + + assertTrue("Invalid " + toString(l), prevEnd <= curStart); + assertTrue("Invalid " + toString(l), curStart <= curEnd); + + if (curStart == curEnd && prevEnd == curStart) + assertTrue("Invalid " + toString(l), prevStart != prevEnd); + + prevStart = curStart; + prevEnd = curEnd; + } + } + private static String toString(RangeTombstone rt) { return String.format("[%d, %d]@%d", i(rt.min), i(rt.max), rt.data.markedForDeleteAt); } + private static String toString(RangeTombstoneList l) + { + StringBuilder sb = new StringBuilder(); + sb.append("{"); + for (RangeTombstone rt : l) + sb.append(" ").append(toString(rt)); + return sb.append(" }").toString(); + } + private static Composite b(int i) { return Util.cellname(i); From 2af8c9da5669f860f1339d789a0f3a0c4f65e5c2 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 10:54:41 +0200 Subject: [PATCH 07/12] Version and licenses for 2.0.10 release --- NEWS.txt | 5 +- .../hadoop/pig/CqlNativeStorage.java | 308 ++++++++++++++++++ 2 files changed, 310 insertions(+), 3 deletions(-) create mode 100644 src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java diff --git a/NEWS.txt b/NEWS.txt index 79212f8ba1..9b521e4dc2 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -72,13 +72,12 @@ Upgrading 2.0.10 -==== +====== New features ------------ - CqlPaginRecordReader and CqlPagingInputFormat have both been removed. Use CqlInputFormat instead. - - If you are using Leveled Compaction, you can now disable doing - size-tiered + - If you are using Leveled Compaction, you can now disable doing size-tiered compaction in L0 by starting Cassandra with -Dcassandra.disable_stcs_in_l0 (see CASSANDRA-6621 for details). - Shuffle and taketoken have been removed. For clusters that choose to diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java new file mode 100644 index 0000000000..6cce4a90a6 --- /dev/null +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java @@ -0,0 +1,308 @@ +/* + * 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.hadoop.pig; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Iterator; +import java.util.Map; + +import org.apache.cassandra.db.BufferCell; +import org.apache.cassandra.db.Cell; +import org.apache.cassandra.db.composites.CellNames; +import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.hadoop.ConfigHelper; +import org.apache.cassandra.hadoop.cql3.CqlConfigHelper; +import org.apache.cassandra.thrift.CfDef; +import org.apache.cassandra.thrift.ColumnDef; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.RecordReader; +import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit; +import org.apache.pig.data.Tuple; +import org.apache.pig.data.TupleFactory; + +import com.datastax.driver.core.Row; + +public class CqlNativeStorage extends CqlStorage +{ + private RecordReader reader; + private String nativePort; + private String nativeCoreConnections; + private String nativeMaxConnections; + private String nativeMinSimultReqs; + private String nativeMaxSimultReqs; + private String nativeConnectionTimeout; + private String nativeReadConnectionTimeout; + private String nativeReceiveBufferSize; + private String nativeSendBufferSize; + private String nativeSolinger; + private String nativeTcpNodelay; + private String nativeReuseAddress; + private String nativeKeepAlive; + private String nativeAuthProvider; + private String nativeSSLTruststorePath; + private String nativeSSLKeystorePath; + private String nativeSSLTruststorePassword; + private String nativeSSLKeystorePassword; + private String nativeSSLCipherSuites; + private String inputCql; + + public CqlNativeStorage() + { + this(1000); + } + + /** @param pageSize limit number of CQL rows to fetch in a thrift request */ + public CqlNativeStorage(int pageSize) + { + super(pageSize); + DEFAULT_INPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlInputFormat"; + } + + public void prepareToRead(RecordReader reader, PigSplit split) + { + this.reader = reader; + } + + /** get next row */ + public Tuple getNext() throws IOException + { + try + { + // load the next pair + if (!reader.nextKeyValue()) + return null; + + CfInfo cfInfo = getCfInfo(loadSignature); + CfDef cfDef = cfInfo.cfDef; + Row row = reader.getCurrentValue(); + Tuple tuple = TupleFactory.getInstance().newTuple(cfDef.column_metadata.size()); + Iterator itera = cfDef.column_metadata.iterator(); + int i = 0; + while (itera.hasNext()) + { + ColumnDef cdef = itera.next(); + ByteBuffer columnValue = row.getBytesUnsafe(ByteBufferUtil.string(cdef.name.duplicate())); + if (columnValue != null) + { + Cell cell = new BufferCell(CellNames.simpleDense(cdef.name), columnValue); + AbstractType validator = getValidatorMap(cfDef).get(cdef.name); + setTupleValue(tuple, i, cqlColumnToObj(cell, cfDef), validator); + } + else + tuple.set(i, null); + i++; + } + return tuple; + } + catch (InterruptedException e) + { + throw new IOException(e.getMessage()); + } + } + + /** set read configuration settings */ + public void setLocation(String location, Job job) throws IOException + { + conf = job.getConfiguration(); + setLocationFromUri(location); + + if (username != null && password != null) + { + ConfigHelper.setInputKeyspaceUserNameAndPassword(conf, username, password); + CqlConfigHelper.setUserNameAndPassword(conf, username, password); + } + if (splitSize > 0) + ConfigHelper.setInputSplitSize(conf, splitSize); + if (partitionerClass!= null) + ConfigHelper.setInputPartitioner(conf, partitionerClass); + if (initHostAddress != null) + ConfigHelper.setInputInitialAddress(conf, initHostAddress); + if (rpcPort != null) + ConfigHelper.setInputRpcPort(conf, rpcPort); + if (nativePort != null) + CqlConfigHelper.setInputNativePort(conf, nativePort); + if (nativeCoreConnections != null) + CqlConfigHelper.setInputCoreConnections(conf, nativeCoreConnections); + if (nativeMaxConnections != null) + CqlConfigHelper.setInputMaxConnections(conf, nativeMaxConnections); + if (nativeMinSimultReqs != null) + CqlConfigHelper.setInputMinSimultReqPerConnections(conf, nativeMinSimultReqs); + if (nativeMaxSimultReqs != null) + CqlConfigHelper.setInputMaxSimultReqPerConnections(conf, nativeMaxSimultReqs); + if (nativeConnectionTimeout != null) + CqlConfigHelper.setInputNativeConnectionTimeout(conf, nativeConnectionTimeout); + if (nativeReadConnectionTimeout != null) + CqlConfigHelper.setInputNativeReadConnectionTimeout(conf, nativeReadConnectionTimeout); + if (nativeReceiveBufferSize != null) + CqlConfigHelper.setInputNativeReceiveBufferSize(conf, nativeReceiveBufferSize); + if (nativeSendBufferSize != null) + CqlConfigHelper.setInputNativeSendBufferSize(conf, nativeSendBufferSize); + if (nativeSolinger != null) + CqlConfigHelper.setInputNativeSolinger(conf, nativeSolinger); + if (nativeTcpNodelay != null) + CqlConfigHelper.setInputNativeTcpNodelay(conf, nativeTcpNodelay); + if (nativeReuseAddress != null) + CqlConfigHelper.setInputNativeReuseAddress(conf, nativeReuseAddress); + if (nativeKeepAlive != null) + CqlConfigHelper.setInputNativeKeepAlive(conf, nativeKeepAlive); + if (nativeAuthProvider != null) + CqlConfigHelper.setInputNativeAuthProvider(conf, nativeAuthProvider); + if (nativeSSLTruststorePath != null) + CqlConfigHelper.setInputNativeSSLTruststorePath(conf, nativeSSLTruststorePath); + if (nativeSSLKeystorePath != null) + CqlConfigHelper.setInputNativeSSLKeystorePath(conf, nativeSSLKeystorePath); + if (nativeSSLTruststorePassword != null) + CqlConfigHelper.setInputNativeSSLTruststorePassword(conf, nativeSSLTruststorePassword); + if (nativeSSLKeystorePassword != null) + CqlConfigHelper.setInputNativeSSLKeystorePassword(conf, nativeSSLKeystorePassword); + if (nativeSSLCipherSuites != null) + CqlConfigHelper.setInputNativeSSLCipherSuites(conf, nativeSSLCipherSuites); + + ConfigHelper.setInputColumnFamily(conf, keyspace, column_family); + setConnectionInformation(); + + CqlConfigHelper.setInputCQLPageRowSize(conf, String.valueOf(pageSize)); + CqlConfigHelper.setInputCql(conf, inputCql); + if (System.getenv(PIG_INPUT_SPLIT_SIZE) != null) + { + try + { + ConfigHelper.setInputSplitSize(conf, Integer.parseInt(System.getenv(PIG_INPUT_SPLIT_SIZE))); + } + catch (NumberFormatException e) + { + throw new IOException("PIG_INPUT_SPLIT_SIZE is not a number", e); + } + } + + if (ConfigHelper.getInputInitialAddress(conf) == null) + throw new IOException("PIG_INPUT_INITIAL_ADDRESS or PIG_INITIAL_ADDRESS environment variable not set"); + if (ConfigHelper.getInputPartitioner(conf) == null) + throw new IOException("PIG_INPUT_PARTITIONER or PIG_PARTITIONER environment variable not set"); + if (loadSignature == null) + loadSignature = location; + + initSchema(loadSignature); + } + + private void setLocationFromUri(String location) throws IOException + { + try + { + if (!location.startsWith("cql://")) + throw new Exception("Bad scheme: " + location); + + String[] urlParts = location.split("\\?"); + if (urlParts.length > 1) + { + Map urlQuery = getQueryMap(urlParts[1]); + + // each page row size + if (urlQuery.containsKey("page_size")) + pageSize = Integer.parseInt(urlQuery.get("page_size")); + + // output prepared statement + if (urlQuery.containsKey("output_query")) + outputQuery = urlQuery.get("output_query"); + + //split size + if (urlQuery.containsKey("split_size")) + splitSize = Integer.parseInt(urlQuery.get("split_size")); + if (urlQuery.containsKey("partitioner")) + partitionerClass = urlQuery.get("partitioner"); + if (urlQuery.containsKey("use_secondary")) + usePartitionFilter = Boolean.parseBoolean(urlQuery.get("use_secondary")); + if (urlQuery.containsKey("init_address")) + initHostAddress = urlQuery.get("init_address"); + + if (urlQuery.containsKey("native_port")) + nativePort = urlQuery.get("native_port"); + if (urlQuery.containsKey("core_conns")) + nativeCoreConnections = urlQuery.get("core_conns"); + if (urlQuery.containsKey("max_conns")) + nativeMaxConnections = urlQuery.get("max_conns"); + if (urlQuery.containsKey("min_simult_reqs")) + nativeMinSimultReqs = urlQuery.get("min_simult_reqs"); + if (urlQuery.containsKey("max_simult_reqs")) + nativeMaxSimultReqs = urlQuery.get("max_simult_reqs"); + if (urlQuery.containsKey("native_timeout")) + nativeConnectionTimeout = urlQuery.get("native_timeout"); + if (urlQuery.containsKey("native_read_timeout")) + nativeReadConnectionTimeout = urlQuery.get("native_read_timeout"); + if (urlQuery.containsKey("rec_buff_size")) + nativeReceiveBufferSize = urlQuery.get("rec_buff_size"); + if (urlQuery.containsKey("send_buff_size")) + nativeSendBufferSize = urlQuery.get("send_buff_size"); + if (urlQuery.containsKey("solinger")) + nativeSolinger = urlQuery.get("solinger"); + if (urlQuery.containsKey("tcp_nodelay")) + nativeTcpNodelay = urlQuery.get("tcp_nodelay"); + if (urlQuery.containsKey("reuse_address")) + nativeReuseAddress = urlQuery.get("reuse_address"); + if (urlQuery.containsKey("keep_alive")) + nativeKeepAlive = urlQuery.get("keep_alive"); + if (urlQuery.containsKey("auth_provider")) + nativeAuthProvider = urlQuery.get("auth_provider"); + if (urlQuery.containsKey("trust_store_path")) + nativeSSLTruststorePath = urlQuery.get("trust_store_path"); + if (urlQuery.containsKey("key_store_path")) + nativeSSLKeystorePath = urlQuery.get("key_store_path"); + if (urlQuery.containsKey("trust_store_password")) + nativeSSLTruststorePassword = urlQuery.get("trust_store_password"); + if (urlQuery.containsKey("key_store_password")) + nativeSSLKeystorePassword = urlQuery.get("key_store_password"); + if (urlQuery.containsKey("cipher_suites")) + nativeSSLCipherSuites = urlQuery.get("cipher_suites"); + if (urlQuery.containsKey("input_cql")) + inputCql = urlQuery.get("input_cql"); + if (urlQuery.containsKey("rpc_port")) + rpcPort = urlQuery.get("rpc_port"); + } + String[] parts = urlParts[0].split("/+"); + String[] credentialsAndKeyspace = parts[1].split("@"); + if (credentialsAndKeyspace.length > 1) + { + String[] credentials = credentialsAndKeyspace[0].split(":"); + username = credentials[0]; + password = credentials[1]; + keyspace = credentialsAndKeyspace[1]; + } + else + { + keyspace = parts[1]; + } + column_family = parts[2]; + } + catch (Exception e) + { + throw new IOException("Expected 'cql://[username:password@]/" + + "[?[page_size=][&columns=][&output_query=]" + + "[&where_clause=][&split_size=][&partitioner=][&use_secondary=true|false]" + + "[&init_address=][&native_port=][&core_conns=]" + + "[&max_conns=][&min_simult_reqs=][&max_simult_reqs=]" + + "[&native_timeout=][&native_read_timeout=][&rec_buff_size=]" + + "[&send_buff_size=][&solinger=][&tcp_nodelay=][&reuse_address=]" + + "[&keep_alive=][&auth_provider=][&trust_store_path=]" + + "[&key_store_path=][&trust_store_password=]" + + "[&key_store_password=][&cipher_suites=][&input_cql=]]': " + e.getMessage()); + } + } + +} From d93b8519100076321228935b853829ac78df44ac Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 8 Aug 2014 17:50:56 +0200 Subject: [PATCH 08/12] Fix build --- .../org/apache/cassandra/hadoop/pig/CqlStorage.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java index 6f17468d9b..3b7b0dff7a 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java @@ -58,11 +58,11 @@ public class CqlStorage extends AbstractCassandraStorage { private static final Logger logger = LoggerFactory.getLogger(CqlStorage.class); private RecordReader, Map> reader; - private RecordWriter, List> writer; + protected RecordWriter, List> writer; - private int pageSize = 1000; + protected int pageSize = 1000; private String columns; - private String outputQuery; + protected String outputQuery; private String whereClause; private boolean hasCompactValueAlias = false; @@ -130,7 +130,7 @@ public class CqlStorage extends AbstractCassandraStorage } /** set the value to the position of the tuple */ - private void setTupleValue(Tuple tuple, int position, Object value, AbstractType validator) throws ExecException + protected void setTupleValue(Tuple tuple, int position, Object value, AbstractType validator) throws ExecException { if (validator instanceof CollectionType) setCollectionTupleValues(tuple, position, value, validator); @@ -184,7 +184,7 @@ public class CqlStorage extends AbstractCassandraStorage } /** convert a cql column to an object */ - private Object cqlColumnToObj(Cell col, CfDef cfDef) throws IOException + protected Object cqlColumnToObj(Cell col, CfDef cfDef) throws IOException { // standard Map validators = getValidatorMap(cfDef); From b2dcaf269a17acaf9755a14f867844efb56f48e7 Mon Sep 17 00:00:00 2001 From: Aleksey Yeschenko Date: Fri, 8 Aug 2014 21:16:55 +0300 Subject: [PATCH 09/12] Fix test/system/test_thrift_server.py --- test/system/test_thrift_server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/system/test_thrift_server.py b/test/system/test_thrift_server.py index dbbc1bfb6b..7655958e40 100644 --- a/test/system/test_thrift_server.py +++ b/test/system/test_thrift_server.py @@ -1222,7 +1222,7 @@ class TestMutations(ThriftTester): def test_describe_keyspace(self): kspaces = client.describe_keyspaces() - assert len(kspaces) == 5, kspaces # ['Keyspace2', 'Keyspace1', 'system', 'system_traces', 'system_auth'] + assert len(kspaces) == 4, kspaces # ['Keyspace2', 'Keyspace1', 'system', 'system_traces'] sysks = client.describe_keyspace("system") assert sysks in kspaces From 74847893c8aa6d736474ab94f219a84cfe13c2ae Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Sat, 9 Aug 2014 13:47:44 +0200 Subject: [PATCH 10/12] Versions for 2.1.0-rc6 --- CHANGES.txt | 2 +- build.xml | 2 +- debian/changelog | 6 ++++++ 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index fff06ae536..31744664a2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,4 @@ -2.1.0-final +2.1.0-rc6 * Include stress yaml example in release and deb (CASSANDRA-7717) * workaround for netty issue causing corrupted data off the wire (CASSANDRA-7695) * cqlsh DESC CLUSTER fails retrieving ring information (CASSANDRA-7687) diff --git a/build.xml b/build.xml index 18fa1ec40a..a76ce366f6 100644 --- a/build.xml +++ b/build.xml @@ -25,7 +25,7 @@ - + diff --git a/debian/changelog b/debian/changelog index 03e308e075..e467737bd3 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (2.1.0~rc6) unstable; urgency=medium + + * New RC release + + -- Sylvain Lebresne Sat, 09 Aug 2014 13:46:39 +0200 + cassandra (2.1.0~rc5) unstable; urgency=medium * New RC release From b55f38be31bc594fcdf6af757e79de2eb7579b05 Mon Sep 17 00:00:00 2001 From: Aleksey Yeschenko Date: Sun, 10 Aug 2014 23:54:13 +0300 Subject: [PATCH 11/12] Fix more of test/system/test_thrift_server.py --- test/system/test_thrift_server.py | 43 ++++++++++++++----------------- 1 file changed, 20 insertions(+), 23 deletions(-) diff --git a/test/system/test_thrift_server.py b/test/system/test_thrift_server.py index 7655958e40..159435653e 100644 --- a/test/system/test_thrift_server.py +++ b/test/system/test_thrift_server.py @@ -24,7 +24,6 @@ from . import thrift_client as client from thrift.Thrift import TApplicationException from ttypes import * -from constants import VERSION def _i64(n): @@ -1236,8 +1235,6 @@ class TestMutations(ThriftTester): assert cf0.comparator_type == "org.apache.cassandra.db.marshal.BytesType" def test_describe(self): - server_version = client.describe_version() - assert server_version == VERSION, (server_version, VERSION) assert client.describe_cluster_name() == 'Test Cluster' def test_describe_ring(self): @@ -1685,13 +1682,13 @@ class TestMutations(ThriftTester): _assert_no_columnpath('key1', ColumnPath(column_family='Counter1', column='c1')) # insert again and this time delete the whole row, check that it is gone - client.add('key1', ColumnParent(column_family='Counter1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) + client.add('key2', ColumnParent(column_family='Counter1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) time.sleep(5) - rv2 = client.get('key1', ColumnPath(column_family='Counter1', column='c1'), ConsistencyLevel.ONE) + rv2 = client.get('key2', ColumnPath(column_family='Counter1', column='c1'), ConsistencyLevel.ONE) assert rv2.counter_column.value == d1 - client.remove_counter('key1', ColumnPath(column_family='Counter1'), ConsistencyLevel.ONE) + client.remove_counter('key2', ColumnPath(column_family='Counter1'), ConsistencyLevel.ONE) time.sleep(5) - _assert_no_columnpath('key1', ColumnPath(column_family='Counter1', column='c1')) + _assert_no_columnpath('key2', ColumnPath(column_family='Counter1', column='c1')) def test_incr_super_remove(self): _set_keyspace('Keyspace1') @@ -1710,13 +1707,13 @@ class TestMutations(ThriftTester): _assert_no_columnpath('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1')) # insert again and this time delete the whole row, check that it is gone - client.add('key1', ColumnParent(column_family='SuperCounter1', super_column='sc1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) + client.add('key2', ColumnParent(column_family='SuperCounter1', super_column='sc1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) time.sleep(5) - rv2 = client.get('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1'), ConsistencyLevel.ONE) + rv2 = client.get('key2', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1'), ConsistencyLevel.ONE) assert rv2.counter_column.value == d1 - client.remove_counter('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1'), ConsistencyLevel.ONE) + client.remove_counter('key2', ColumnPath(column_family='SuperCounter1', super_column='sc1'), ConsistencyLevel.ONE) time.sleep(5) - _assert_no_columnpath('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1')) + _assert_no_columnpath('key2', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1')) def test_incr_decr_standard_remove(self): _set_keyspace('Keyspace1') @@ -1735,13 +1732,13 @@ class TestMutations(ThriftTester): _assert_no_columnpath('key1', ColumnPath(column_family='Counter1', column='c1')) # insert again and this time delete the whole row, check that it is gone - client.add('key1', ColumnParent(column_family='Counter1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) + client.add('key2', ColumnParent(column_family='Counter1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) time.sleep(5) - rv2 = client.get('key1', ColumnPath(column_family='Counter1', column='c1'), ConsistencyLevel.ONE) + rv2 = client.get('key2', ColumnPath(column_family='Counter1', column='c1'), ConsistencyLevel.ONE) assert rv2.counter_column.value == d1 - client.remove_counter('key1', ColumnPath(column_family='Counter1'), ConsistencyLevel.ONE) + client.remove_counter('key2', ColumnPath(column_family='Counter1'), ConsistencyLevel.ONE) time.sleep(5) - _assert_no_columnpath('key1', ColumnPath(column_family='Counter1', column='c1')) + _assert_no_columnpath('key2', ColumnPath(column_family='Counter1', column='c1')) def test_incr_decr_super_remove(self): _set_keyspace('Keyspace1') @@ -1760,13 +1757,13 @@ class TestMutations(ThriftTester): _assert_no_columnpath('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1')) # insert again and this time delete the whole row, check that it is gone - client.add('key1', ColumnParent(column_family='SuperCounter1', super_column='sc1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) + client.add('key2', ColumnParent(column_family='SuperCounter1', super_column='sc1'), CounterColumn('c1', d1), ConsistencyLevel.ONE) time.sleep(5) - rv2 = client.get('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1'), ConsistencyLevel.ONE) + rv2 = client.get('key2', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1'), ConsistencyLevel.ONE) assert rv2.counter_column.value == d1 - client.remove_counter('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1'), ConsistencyLevel.ONE) + client.remove_counter('key2', ColumnPath(column_family='SuperCounter1', super_column='sc1'), ConsistencyLevel.ONE) time.sleep(5) - _assert_no_columnpath('key1', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1')) + _assert_no_columnpath('key2', ColumnPath(column_family='SuperCounter1', super_column='sc1', column='c1')) def test_incr_decr_standard_batch_add(self): _set_keyspace('Keyspace1') @@ -1809,21 +1806,21 @@ class TestMutations(ThriftTester): _assert_no_columnpath('key1', ColumnPath(column_family='Counter1', column='c1')) # insert again and this time delete the whole row, check that it is gone - update_map = {'key1': {'Counter1': [ + update_map = {'key2': {'Counter1': [ Mutation(column_or_supercolumn=ColumnOrSuperColumn(counter_column=CounterColumn('c1', d1))), Mutation(column_or_supercolumn=ColumnOrSuperColumn(counter_column=CounterColumn('c1', d2))), ]}} client.batch_mutate(update_map, ConsistencyLevel.ONE) time.sleep(5) - rv2 = client.get('key1', ColumnPath(column_family='Counter1', column='c1'), ConsistencyLevel.ONE) + rv2 = client.get('key2', ColumnPath(column_family='Counter1', column='c1'), ConsistencyLevel.ONE) assert rv2.counter_column.value == d1+d2 - update_map = {'key1': {'Counter1': [ + update_map = {'key2': {'Counter1': [ Mutation(deletion=Deletion()), ]}} client.batch_mutate(update_map, ConsistencyLevel.ONE) time.sleep(5) - _assert_no_columnpath('key1', ColumnPath(column_family='Counter1', column='c1')) + _assert_no_columnpath('key2', ColumnPath(column_family='Counter1', column='c1')) def test_incr_decr_standard_slice(self): _set_keyspace('Keyspace1') From a771f60dbb43ba4b86afb5afe501620a816b1b3d Mon Sep 17 00:00:00 2001 From: Aleksey Yeschenko Date: Mon, 11 Aug 2014 00:26:36 +0300 Subject: [PATCH 12/12] Fix even more of test/system/test_thrift_server.py --- test/system/test_thrift_server.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/system/test_thrift_server.py b/test/system/test_thrift_server.py index bef08a2bbe..5f094751c2 100644 --- a/test/system/test_thrift_server.py +++ b/test/system/test_thrift_server.py @@ -630,7 +630,7 @@ class TestMutations(ThriftTester): for column in columns: client.insert('key', ColumnParent('Standard1'), column, ConsistencyLevel.ONE) - d = Deletion(1, predicate=SlicePredicate(slice_range=SliceRange(start='c2', finish='c5'))) + d = Deletion(1, predicate=SlicePredicate(slice_range=SliceRange(start='c2', finish='c4'))) client.batch_mutate({'key': {'Standard1' : [Mutation(deletion=d)]}}, ConsistencyLevel.ONE) _assert_columnpath_exists('key', ColumnPath('Standard1', column='c1')) @@ -677,7 +677,7 @@ class TestMutations(ThriftTester): for column in columns: client.insert('key', ColumnParent('Super1', 'sc1'), column, ConsistencyLevel.ONE) - r = SliceRange(start=_i64(2), finish=_i64(5)) + r = SliceRange(start=_i64(2), finish=_i64(4)) d = Deletion(1, super_column='sc1', predicate=SlicePredicate(slice_range=r)) client.batch_mutate({'key': {'Super1' : [Mutation(deletion=d)]}}, ConsistencyLevel.ONE)