Merge branch 'cassandra-3.0' into cassandra-3.11
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/08a515dc Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/08a515dc Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/08a515dc Branch: refs/heads/trunk Commit: 08a515dc7440db1530987a480afb27f5faaf8e0f Parents: 191ad7b 4e23c9e Author: Aleksey Yeshchenko <[email protected]> Authored: Wed Jun 20 16:03:25 2018 +0100 Committer: Aleksey Yeshchenko <[email protected]> Committed: Wed Jun 20 16:03:25 2018 +0100 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../org/apache/cassandra/db/ReadCommand.java | 44 +- .../db/rows/UnfilteredRowIterators.java | 26 + .../cassandra/db/transform/RTBoundCloser.java | 110 +++++ .../db/transform/RTBoundValidator.java | 106 ++++ .../apache/cassandra/service/DataResolver.java | 21 +- .../db/transform/RTTransformationsTest.java | 483 +++++++++++++++++++ 7 files changed, 760 insertions(+), 31 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/08a515dc/CHANGES.txt ---------------------------------------------------------------------- diff --cc CHANGES.txt index e807340,aeeb0ae..d9c62ea --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,22 -1,5 +1,23 @@@ -3.0.17 +3.11.3 + * Remove BTree.Builder Recycler to reduce memory usage (CASSANDRA-13929) + * Reduce nodetool GC thread count (CASSANDRA-14475) + * Fix New SASI view creation during Index Redistribution (CASSANDRA-14055) + * Remove string formatting lines from BufferPool hot path (CASSANDRA-14416) + * Update metrics to 3.1.5 (CASSANDRA-12924) + * Detect OpenJDK jvm type and architecture (CASSANDRA-12793) + * Don't use guava collections in the non-system keyspace jmx attributes (CASSANDRA-12271) + * Allow existing nodes to use all peers in shadow round (CASSANDRA-13851) + * Fix cqlsh to read connection.ssl cqlshrc option again (CASSANDRA-14299) + * Downgrade log level to trace for CommitLogSegmentManager (CASSANDRA-14370) + * CQL fromJson(null) throws NullPointerException (CASSANDRA-13891) + * Serialize empty buffer as empty string for json output format (CASSANDRA-14245) + * Allow logging implementation to be interchanged for embedded testing (CASSANDRA-13396) + * SASI tokenizer for simple delimiter based entries (CASSANDRA-14247) + * Fix Loss of digits when doing CAST from varint/bigint to decimal (CASSANDRA-14170) + * RateBasedBackPressure unnecessarily invokes a lock on the Guava RateLimiter (CASSANDRA-14163) + * Fix wildcard GROUP BY queries (CASSANDRA-14209) +Merged from 3.0: + * Always close RT markers returned by ReadCommand#executeLocally() (CASSANDRA-14515) * Reverse order queries with range tombstones can cause data loss (CASSANDRA-14513) * Fix regression of lagging commitlog flush log message (CASSANDRA-14451) * Add Missing dependencies in pom-all (CASSANDRA-14422) http://git-wip-us.apache.org/repos/asf/cassandra/blob/08a515dc/src/java/org/apache/cassandra/db/ReadCommand.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/db/ReadCommand.java index c8b256a,f8a0795..8bcfcbc --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@@ -31,11 -31,10 +31,13 @@@ import org.slf4j.LoggerFactory import org.apache.cassandra.config.*; import org.apache.cassandra.cql3.Operator; import org.apache.cassandra.db.filter.*; +import org.apache.cassandra.db.monitoring.ApproximateTime; +import org.apache.cassandra.db.monitoring.MonitorableImpl; import org.apache.cassandra.db.partitions.*; import org.apache.cassandra.db.rows.*; + import org.apache.cassandra.db.transform.RTBoundCloser; + import org.apache.cassandra.db.transform.RTBoundValidator; +import org.apache.cassandra.db.transform.StoppingTransformation; import org.apache.cassandra.db.transform.Transformation; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.index.Index; @@@ -413,14 -407,12 +419,13 @@@ public abstract class ReadCommand exten Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.ksName, cfs.metadata.cfName, index.getIndexMetadata().name); } - UnfilteredPartitionIterator resultIterator = searcher == null - ? queryStorage(cfs, executionController) - : searcher.search(executionController); - UnfilteredPartitionIterator iterator = (null == searcher) ? queryStorage(cfs, orderGroup) : searcher.search(orderGroup); ++ UnfilteredPartitionIterator iterator = (null == searcher) ? queryStorage(cfs, executionController) : searcher.search(executionController); try { - resultIterator = withStateTracking(resultIterator); - resultIterator = withMetricsRecording(withoutPurgeableTombstones(resultIterator, cfs), cfs.metric, startTimeNanos); ++ iterator = withStateTracking(iterator); + iterator = withoutPurgeableTombstones(iterator, cfs); + iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos); // If we've used a 2ndary index, we know the result already satisfy the primary expression used, so // no point in checking it again. http://git-wip-us.apache.org/repos/asf/cassandra/blob/08a515dc/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/08a515dc/src/java/org/apache/cassandra/db/transform/RTBoundCloser.java ---------------------------------------------------------------------- diff --cc src/java/org/apache/cassandra/db/transform/RTBoundCloser.java index 0000000,11f0344..413ea71 mode 000000,100644..100644 --- a/src/java/org/apache/cassandra/db/transform/RTBoundCloser.java +++ b/src/java/org/apache/cassandra/db/transform/RTBoundCloser.java @@@ -1,0 -1,110 +1,110 @@@ + /* + * 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.transform; + + import org.apache.cassandra.config.CFMetaData; + import org.apache.cassandra.db.Clustering; + import org.apache.cassandra.db.DeletionTime; -import org.apache.cassandra.db.ReadOrderGroup; ++import org.apache.cassandra.db.ReadExecutionController; + import org.apache.cassandra.db.rows.*; + + /** + * A transformation that appends an RT bound marker to row iterators in case they don't have one. + * - * This used to happen, for example, in {@link org.apache.cassandra.db.ReadCommand#executeLocally(ReadOrderGroup)}, ++ * This used to happen, for example, in {@link org.apache.cassandra.db.ReadCommand#executeLocally(ReadExecutionController)} + * if {@link org.apache.cassandra.db.filter.DataLimits} stopped the iterator on a live row that was enclosed in an + * older RT. + * + * If we don't do this, and send a response without the closing bound, we can break read/short read protection read + * isolation, and potentially cause data loss. + * + * See CASSANDRA-14515 for context. + */ + public final class RTBoundCloser extends Transformation<UnfilteredRowIterator> + { + @Override + public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition) + { + RowsTransformation transformation = new RowsTransformation(partition); + return Transformation.apply(MoreRows.extend(partition, transformation, partition.columns()), transformation); + } + + private final static class RowsTransformation extends Transformation implements MoreRows<UnfilteredRowIterator> + { + private final UnfilteredRowIterator partition; + + private Clustering lastRowClustering; + private DeletionTime openMarkerDeletionTime; + + private RowsTransformation(UnfilteredRowIterator partition) + { + this.partition = partition; + } + + @Override + public Row applyToRow(Row row) + { + lastRowClustering = row.clustering(); + return row; + } + + @Override + public RangeTombstoneMarker applyToMarker(RangeTombstoneMarker marker) + { + openMarkerDeletionTime = + marker.isOpen(partition.isReverseOrder()) ? marker.openDeletionTime(partition.isReverseOrder()) : null; + lastRowClustering = null; + return marker; + } + + @Override + public UnfilteredRowIterator moreContents() + { + // there is no open RT in the stream - nothing for us to do + if (null == openMarkerDeletionTime) + return null; + + /* + * there *is* an open RT in the stream, but there have been no rows after the opening bound - this must + * never happen in scenarios where RTBoundCloser is meant to be used; the last encountered clustering + * should be either a closing bound marker - if the iterator was exhausted fully - or a live row - if + * DataLimits stopped it short in the middle of an RT. + */ + if (null == lastRowClustering) + { + CFMetaData metadata = partition.metadata(); + String message = + String.format("UnfilteredRowIterator for %s.%s has an open RT bound as its last item", metadata.ksName, metadata.cfName); + throw new IllegalStateException(message); + } + + // create an artificial inclusive closing RT bound with bound matching last seen row's clustering + RangeTombstoneBoundMarker closingBound = + RangeTombstoneBoundMarker.inclusiveClose(partition.isReverseOrder(), lastRowClustering.getRawValues(), openMarkerDeletionTime); + + return UnfilteredRowIterators.singleton(closingBound, + partition.metadata(), + partition.partitionKey(), + partition.partitionLevelDeletion(), + partition.columns(), + partition.staticRow(), + partition.isReverseOrder(), + partition.stats()); + } + } + } http://git-wip-us.apache.org/repos/asf/cassandra/blob/08a515dc/src/java/org/apache/cassandra/service/DataResolver.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/08a515dc/test/unit/org/apache/cassandra/db/transform/RTTransformationsTest.java ---------------------------------------------------------------------- diff --cc test/unit/org/apache/cassandra/db/transform/RTTransformationsTest.java index 0000000,832c5a3..ee91135 mode 000000,100644..100644 --- a/test/unit/org/apache/cassandra/db/transform/RTTransformationsTest.java +++ b/test/unit/org/apache/cassandra/db/transform/RTTransformationsTest.java @@@ -1,0 -1,482 +1,483 @@@ + /* + * 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.transform; + + import java.nio.ByteBuffer; + import java.util.*; + + import org.junit.Before; + import org.junit.Test; + + import com.google.common.collect.Iterators; + + import org.apache.cassandra.config.CFMetaData; + import org.apache.cassandra.db.*; + import org.apache.cassandra.db.ClusteringPrefix.Kind; + import org.apache.cassandra.db.marshal.AbstractType; + import org.apache.cassandra.db.marshal.UTF8Type; + import org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator; + import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; + import org.apache.cassandra.db.rows.*; + import org.apache.cassandra.dht.Murmur3Partitioner; + import org.apache.cassandra.utils.FBUtilities; + + import static org.junit.Assert.assertEquals; + import static org.junit.Assert.assertFalse; + import static org.junit.Assert.assertTrue; + + import static org.apache.cassandra.utils.ByteBufferUtil.bytes; + + public final class RTTransformationsTest + { + private static final String KEYSPACE = "RTBoundCloserTest"; + private static final String TABLE = "table"; + + private final int nowInSec = FBUtilities.nowInSeconds(); + + private CFMetaData metadata; + private DecoratedKey key; + + @Before + public void setUp() + { + metadata = + CFMetaData.Builder + .create(KEYSPACE, TABLE) + .addPartitionKey("pk", UTF8Type.instance) + .addClusteringColumn("ck0", UTF8Type.instance) + .addClusteringColumn("ck1", UTF8Type.instance) + .addClusteringColumn("ck2", UTF8Type.instance) ++ .withPartitioner(Murmur3Partitioner.instance) + .build(); + key = Murmur3Partitioner.instance.decorateKey(bytes("key")); + } + + @Test + public void testAddsNothingWhenAlreadyClosed() + { + UnfilteredPartitionIterator original = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + ); + + UnfilteredPartitionIterator extended = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + ); + extended = Transformation.apply(extended, new RTBoundCloser()); + assertIteratorsEqual(original, extended); + } + + @Test + public void testAddsNothingWhenAlreadyClosedInReverseOrder() + { + UnfilteredPartitionIterator original = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + ); + + UnfilteredPartitionIterator extended = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + ); + extended = Transformation.apply(extended, new RTBoundCloser()); + assertIteratorsEqual(original, extended); + } + + @Test + public void testClosesUnclosedBound() + { + UnfilteredPartitionIterator original = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + ); + UnfilteredPartitionIterator extended = Transformation.apply(original, new RTBoundCloser()); + + UnfilteredPartitionIterator expected = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1", "") + ); + assertIteratorsEqual(expected, extended); + } + + @Test + public void testClosesUnclosedBoundary() + { + UnfilteredPartitionIterator original = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a") + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 1, "a", "0") + , row(2, "a", "1", "") + ); + UnfilteredPartitionIterator extended = Transformation.apply(original, new RTBoundCloser()); + + UnfilteredPartitionIterator expected = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a") + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 1, "a", "0") + , row(2, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 1, "a", "1", "") + ); + assertIteratorsEqual(expected, extended); + } + + @Test + public void testClosesUnclosedBoundInReverseOrder() + { + UnfilteredPartitionIterator original = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + ); + UnfilteredPartitionIterator extended = Transformation.apply(original, new RTBoundCloser()); + + UnfilteredPartitionIterator expected = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1", "") + ); + assertIteratorsEqual(expected, extended); + } + + @Test + public void testClosesUnclosedBoundaryInReverseOrder() + { + UnfilteredPartitionIterator original = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a") + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 1, 0, "a", "1") + , row(2, "a", "0", "") + ); + UnfilteredPartitionIterator extended = Transformation.apply(original, new RTBoundCloser()); + + UnfilteredPartitionIterator expected = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a") + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 1, 0, "a", "1") + , row(2, "a", "0", "") + , bound(Kind.INCL_START_BOUND, 1, "a", "0", "") + ); + + assertIteratorsEqual(expected, extended); + } + + @Test + public void testFailsWithoutSeeingRows() + { + UnfilteredPartitionIterator iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a") + ); + iterator = Transformation.apply(iterator, new RTBoundCloser()); + assertThrowsISEIterated(iterator); + } + + @Test + public void testValidatesLegalBounds() + { + UnfilteredPartitionIterator iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + + , bound(Kind.INCL_START_BOUND, 0, "a", "2") + , row(1, "a", "2", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "2") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + drain(iterator); + } + + @Test + public void testValidatesLegalBoundsInReverseOrder() + { + UnfilteredPartitionIterator iterator = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "2") + , row(1, "a", "2", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "2") + + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + drain(iterator); + } + + @Test + public void testValidatesLegalBoundaries() + { + UnfilteredPartitionIterator iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a") + + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 1, "a", "1") + , row(2, "a", "1", "") + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 1, 0, "a", "1") + + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 2, "a", "2") + , row(3, "a", "2", "") + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 2, 0, "a", "2") + + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 3, "a", "3") + , row(4, "a", "3", "") + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 3, 0, "a", "3") + + , bound(Kind.INCL_END_BOUND, 0, "a") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + drain(iterator); + } + + @Test + public void testValidatesLegalBoundariesInReverseOrder() + { + UnfilteredPartitionIterator iterator = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a") + + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 3, 0, "a", "3") + , row(4, "a", "3", "") + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 3, "a", "3") + + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 2, 0, "a", "2") + , row(3, "a", "2", "") + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 2, "a", "2") + + , boundary(Kind.INCL_END_EXCL_START_BOUNDARY, 1, 0, "a", "1") + , row(2, "a", "1", "") + , boundary(Kind.EXCL_END_INCL_START_BOUNDARY, 0, 1, "a", "1") + + , bound(Kind.INCL_START_BOUND, 0, "a") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + drain(iterator); + } + + @Test + public void testComplainsAboutMismatchedTimestamps() + { + UnfilteredPartitionIterator iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 1, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + } + + @Test + public void testComplainsAboutMismatchedTimestampsInReverseOrder() + { + UnfilteredPartitionIterator iterator = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 1, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + } + + @Test + public void testComplainsAboutInvalidSequence() + { + // duplicated start bound + UnfilteredPartitionIterator iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + + // duplicated end bound + iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + + // absent open bound + iterator = iter(false + , row(1, "a", "1", "") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + + // absent end bound + iterator = iter(false + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + } + + @Test + public void testComplainsAboutInvalidSequenceInReveseOrder() + { + // duplicated start bound + UnfilteredPartitionIterator iterator = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + + // duplicated end bound + iterator = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + + // absent open bound + iterator = iter(true + , bound(Kind.INCL_END_BOUND, 0, "a", "1") + , row(1, "a", "1", "") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + + // absent end bound + iterator = iter(true + , row(1, "a", "1", "") + , bound(Kind.INCL_START_BOUND, 0, "a", "1") + ); + iterator = Transformation.apply(iterator, new RTBoundValidator(true)); + assertThrowsISEIterated(iterator); + } + + private RangeTombstoneBoundMarker bound(ClusteringPrefix.Kind kind, long timestamp, Object... clusteringValues) + { + ByteBuffer[] clusteringByteBuffers = new ByteBuffer[clusteringValues.length]; + for (int i = 0; i < clusteringValues.length; i++) + clusteringByteBuffers[i] = decompose(metadata.clusteringColumns().get(i).type, clusteringValues[i]); + - return new RangeTombstoneBoundMarker(new RangeTombstone.Bound(kind, clusteringByteBuffers), new DeletionTime(timestamp, nowInSec)); ++ return new RangeTombstoneBoundMarker(ClusteringBound.create(kind, clusteringByteBuffers), new DeletionTime(timestamp, nowInSec)); + } + + private RangeTombstoneBoundaryMarker boundary(ClusteringPrefix.Kind kind, long closeTimestamp, long openTimestamp, Object... clusteringValues) + { + ByteBuffer[] clusteringByteBuffers = new ByteBuffer[clusteringValues.length]; + for (int i = 0; i < clusteringValues.length; i++) + clusteringByteBuffers[i] = decompose(metadata.clusteringColumns().get(i).type, clusteringValues[i]); + - return new RangeTombstoneBoundaryMarker(new RangeTombstone.Bound(kind, clusteringByteBuffers), ++ return new RangeTombstoneBoundaryMarker(ClusteringBoundary.create(kind, clusteringByteBuffers), + new DeletionTime(closeTimestamp, nowInSec), + new DeletionTime(openTimestamp, nowInSec)); + } + + private Row row(long timestamp, Object... clusteringValues) + { + ByteBuffer[] clusteringByteBuffers = new ByteBuffer[clusteringValues.length]; + for (int i = 0; i < clusteringValues.length; i++) + clusteringByteBuffers[i] = decompose(metadata.clusteringColumns().get(i).type, clusteringValues[i]); + - return BTreeRow.noCellLiveRow(new Clustering(clusteringByteBuffers), LivenessInfo.create(metadata, timestamp, nowInSec)); ++ return BTreeRow.noCellLiveRow(Clustering.make(clusteringByteBuffers), LivenessInfo.create(timestamp, nowInSec)); + } + + @SuppressWarnings("unchecked") + private static <T> ByteBuffer decompose(AbstractType<?> type, T value) + { + return ((AbstractType<T>) type).decompose(value); + } + + private UnfilteredPartitionIterator iter(boolean isReversedOrder, Unfiltered... unfiltereds) + { + Iterator<Unfiltered> iterator = Iterators.forArray(unfiltereds); + + UnfilteredRowIterator rowIter = + new AbstractUnfilteredRowIterator(metadata, + key, + DeletionTime.LIVE, + metadata.partitionColumns(), + Rows.EMPTY_STATIC_ROW, + isReversedOrder, + EncodingStats.NO_STATS) + { + protected Unfiltered computeNext() + { + return iterator.hasNext() ? iterator.next() : endOfData(); + } + }; + + return new SingletonUnfilteredPartitionIterator(rowIter, false); + } + + private void assertIteratorsEqual(UnfilteredPartitionIterator iter1, UnfilteredPartitionIterator iter2) + { + while (iter1.hasNext()) + { + assertTrue(iter2.hasNext()); + + try (UnfilteredRowIterator partition1 = iter1.next()) + { + try (UnfilteredRowIterator partition2 = iter2.next()) + { + assertIteratorsEqual(partition1, partition2); + } + } + } + assertFalse(iter2.hasNext()); + } + + private void assertIteratorsEqual(UnfilteredRowIterator iter1, UnfilteredRowIterator iter2) + { + while (iter1.hasNext()) + { + assertTrue(iter2.hasNext()); + + assertEquals(iter1.next(), iter2.next()); + } + assertFalse(iter2.hasNext()); + } + + private void assertThrowsISEIterated(UnfilteredPartitionIterator iterator) + { + Throwable t = null; + try + { + drain(iterator); + } + catch (Throwable e) + { + t = e; + } + assertTrue(t instanceof IllegalStateException); + } + + private void drain(UnfilteredPartitionIterator iter) + { + while (iter.hasNext()) + { + try (UnfilteredRowIterator partition = iter.next()) + { + while (partition.hasNext()) + partition.next(); + } + } + } + } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
