diff --git a/CHANGES.txt b/CHANGES.txt index 53e1e6fb21..ba1e72b418 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -37,6 +37,9 @@ Merged from 2.0: * Fix potential paging bug with deleted columns (CASSANDRA-6748) * Catch memtable flush exceptions during shutdown (CASSANDRA-6735) * Fix upgradesstables NPE for non-CF-based indexes (CASSANDRA-6645) + * Fix UPDATE updating PRIMARY KEY columns implicitly (CASSANDRA-6782) + * Fix IllegalArgumentException when updating from 1.2 with SuperColumns + (CASSANDRA-6733) 2.1.0-beta1 diff --git a/conf/cassandra-env.sh b/conf/cassandra-env.sh index 20f26dafff..6b260996f3 100644 --- a/conf/cassandra-env.sh +++ b/conf/cassandra-env.sh @@ -168,6 +168,9 @@ then JVM_OPTS="$JVM_OPTS -javaagent:$CASSANDRA_HOME/lib/jamm-0.2.6.jar" fi +# some JVMs will fill up their heap when accessed via JMX, see CASSANDRA-6541 +JVM_OPTS="$JVM_OPTS -XX:+CMSClassUnloadingEnabled" + # enable thread priorities, primarily so we can give periodic tasks # a lower priority to avoid interfering with client workload JVM_OPTS="$JVM_OPTS -XX:+UseThreadPriorities" diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java index 984f4df672..85ca069860 100644 --- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java @@ -51,17 +51,19 @@ public class UpdateStatement extends ModificationStatement throws InvalidRequestException { // Inserting the CQL row marker (see #4361) - // We always need to insert a marker, because of the following situation: + // We always need to insert a marker for INSERT, because of the following situation: // CREATE TABLE t ( k int PRIMARY KEY, c text ); // INSERT INTO t(k, c) VALUES (1, 1) // DELETE c FROM t WHERE k = 1; // SELECT * FROM t; - // The last query should return one row (but with c == null). Adding - // the marker with the insert make sure the semantic is correct (while making sure a - // 'DELETE FROM t WHERE k = 1' does remove the row entirely) + // The last query should return one row (but with c == null). Adding the marker with the insert make sure + // the semantic is correct (while making sure a 'DELETE FROM t WHERE k = 1' does remove the row entirely) + // + // We do not insert the marker for UPDATE however, as this amount to updating the columns in the WHERE + // clause which is inintuitive (#6782) // // We never insert markers for Super CF as this would confuse the thrift side. - if (cfm.isCQL3Table() && !prefix.isStatic()) + if (type == StatementType.INSERT && cfm.isCQL3Table() && !prefix.isStatic()) cf.addColumn(params.makeColumn(cfm.comparator.rowMarker(prefix), ByteBufferUtil.EMPTY_BYTE_BUFFER)); List updates = getOperations(); diff --git a/src/java/org/apache/cassandra/db/SuperColumns.java b/src/java/org/apache/cassandra/db/SuperColumns.java index b854856e24..46c55777b4 100644 --- a/src/java/org/apache/cassandra/db/SuperColumns.java +++ b/src/java/org/apache/cassandra/db/SuperColumns.java @@ -129,6 +129,11 @@ public class SuperColumns return new SimpleDenseCellNameType(type.subtype(1)); } + public static CellNameType scNameType(CellNameType type) + { + return new SimpleDenseCellNameType(type.subtype(0)); + } + public static AbstractType getComparatorFor(CFMetaData metadata, ByteBuffer superColumn) { return getComparatorFor(metadata, superColumn != null); diff --git a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java index 2a596ea065..9c1fece671 100644 --- a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java +++ b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java @@ -26,6 +26,7 @@ import com.google.common.collect.AbstractIterator; import org.apache.cassandra.db.*; import org.apache.cassandra.db.composites.CellNameType; +import org.apache.cassandra.db.composites.CellNames; import org.apache.cassandra.db.composites.Composite; import org.apache.cassandra.db.filter.ColumnSlice; import org.apache.cassandra.io.sstable.CorruptSSTableException; @@ -178,6 +179,34 @@ class IndexedSliceReader extends AbstractIterator implements OnDiskA } } + static int indexFor(SSTableReader sstable, Composite name, List indexes, CellNameType comparator, boolean reversed, int startIdx) + { + // If it's a super CF and the sstable is from the old format, then the index will contain old format info, i.e. non composite + // SC names. So we need to 1) use only the SC name part of the comparator and 2) extract only that part from 'name' + if (sstable.metadata.isSuper() && sstable.descriptor.version.hasSuperColumns) + { + CellNameType scComparator = SuperColumns.scNameType(comparator); + Composite scName = CellNames.compositeDense(SuperColumns.scName(name)); + return IndexHelper.indexFor(scName, indexes, scComparator, reversed, startIdx); + } + return IndexHelper.indexFor(name, indexes, comparator, reversed, startIdx); + } + + static Composite forIndexComparison(SSTableReader sstable, Composite name) + { + // See indexFor above. + return sstable.metadata.isSuper() && sstable.descriptor.version.hasSuperColumns + ? CellNames.compositeDense(SuperColumns.scName(name)) + : name; + } + + static CellNameType comparatorForIndex(SSTableReader sstable, CellNameType comparator) + { + return sstable.metadata.isSuper() && sstable.descriptor.version.hasSuperColumns + ? SuperColumns.scNameType(comparator) + : comparator; + } + private abstract class BlockFetcher { protected int currentSliceIdx; @@ -218,16 +247,22 @@ class IndexedSliceReader extends AbstractIterator implements OnDiskA return !start.isEmpty() && comparator.compare(name, start) < 0; } + protected boolean isIndexEntryBeforeSliceStart(Composite name) + { + Composite start = currentStart(); + return !start.isEmpty() && comparatorForIndex(sstable, comparator).compare(name, forIndexComparison(sstable, start)) < 0; + } + protected boolean isColumnBeforeSliceFinish(OnDiskAtom column) { Composite finish = currentFinish(); return finish.isEmpty() || comparator.compare(column.name(), finish) <= 0; } - protected boolean isAfterSliceFinish(Composite name) + protected boolean isIndexEntryAfterSliceFinish(Composite name) { Composite finish = currentFinish(); - return !finish.isEmpty() && comparator.compare(name, finish) > 0; + return !finish.isEmpty() && comparatorForIndex(sstable, comparator).compare(name, forIndexComparison(sstable, finish)) > 0; } } @@ -258,7 +293,7 @@ class IndexedSliceReader extends AbstractIterator implements OnDiskA { while (++currentSliceIdx < slices.length) { - nextIndexIdx = IndexHelper.indexFor(slices[currentSliceIdx].start, indexes, comparator, reversed, nextIndexIdx); + nextIndexIdx = indexFor(sstable, slices[currentSliceIdx].start, indexes, comparator, reversed, nextIndexIdx); if (nextIndexIdx < 0 || nextIndexIdx >= indexes.size()) // no index block for that slice continue; @@ -267,12 +302,12 @@ class IndexedSliceReader extends AbstractIterator implements OnDiskA IndexInfo info = indexes.get(nextIndexIdx); if (reversed) { - if (!isBeforeSliceStart(info.lastName)) + if (!isIndexEntryBeforeSliceStart(info.lastName)) return true; } else { - if (!isAfterSliceFinish(info.firstName)) + if (!isIndexEntryAfterSliceFinish(info.firstName)) return true; } } diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java index 2eb6745a64..374dedb1ec 100644 --- a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java +++ b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java @@ -186,13 +186,15 @@ public class SSTableNamesIterator extends AbstractIterator implement int lastIndexIdx = -1; for (CellName name : columnNames) { - int index = IndexHelper.indexFor(name, indexList, comparator, false, lastIndexIdx); + int index = IndexedSliceReader.indexFor(sstable, name, indexList, comparator, false, lastIndexIdx); if (index < 0 || index == indexList.size()) continue; IndexHelper.IndexInfo indexInfo = indexList.get(index); // Check the index block does contain the column names and that we haven't inserted this block yet. - if (comparator.compare(name, indexInfo.firstName) < 0 || index == lastIndexIdx) + if (IndexedSliceReader.comparatorForIndex(sstable, comparator).compare(IndexedSliceReader.forIndexComparison(sstable, name), indexInfo.firstName) < 0 + || index == lastIndexIdx) continue; + ranges.add(indexInfo); lastIndexIdx = index; } diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java index 99bbd85ea3..436b6dcfc9 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java @@ -230,7 +230,7 @@ public class CompressedSequentialWriter extends SequentialWriter // truncate data and index file truncate(chunkOffset); - metadataWriter.resetAndTruncate(realMark.nextChunkIndex); + metadataWriter.resetAndTruncate(realMark.nextChunkIndex - 1); } /** diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java index 587bf0ab02..c330c8808b 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java @@ -145,7 +145,7 @@ public class SSTableLoader implements StreamEventHandler return stream(Collections.emptySet()); } - public StreamResultFuture stream(Set toIgnore) + public StreamResultFuture stream(Set toIgnore, StreamEventHandler... listeners) { client.init(keyspace); outputHandler.output("Established connection to initial hosts"); @@ -176,9 +176,8 @@ public class SSTableLoader implements StreamEventHandler plan.transferFiles(remote, streamingDetails.get(remote)); } - StreamResultFuture bulkResult = plan.execute(); - bulkResult.addEventListener(this); - return bulkResult; + plan.listeners(this, listeners); + return plan.execute(); } public void onSuccess(StreamState finalState) {} diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java index dcbd5ff392..2e1adcb88a 100644 --- a/src/java/org/apache/cassandra/repair/RepairJob.java +++ b/src/java/org/apache/cassandra/repair/RepairJob.java @@ -19,14 +19,13 @@ package org.apache.cassandra.repair; import java.net.InetAddress; import java.util.*; -import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Condition; import com.google.common.util.concurrent.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; @@ -51,12 +50,13 @@ public class RepairJob private final List trees = new ArrayList<>(); // once all responses are received, each tree is compared with each other, and differencer tasks // are submitted. the job is done when all differencers are complete. - private final Set differencers = new HashSet<>(); private final ListeningExecutorService taskExecutor; private final Condition requestsSent = new SimpleCondition(); private int gcBefore = -1; private volatile boolean failed = false; + /* Count down as sync completes */ + private AtomicInteger waitForSync; /** * Create repair job to run on specific columnfamily @@ -172,7 +172,7 @@ public class RepairJob public void submitDifferencers() { assert !failed; - + List differencers = new ArrayList<>(); // We need to difference all trees one against another for (int i = 0; i < trees.size() - 1; ++i) { @@ -183,21 +183,20 @@ public class RepairJob Differencer differencer = new Differencer(desc, r1, r2); differencers.add(differencer); logger.debug("Queueing comparison {}", differencer); - taskExecutor.submit(differencer); } } + waitForSync = new AtomicInteger(differencers.size()); + for (Differencer differencer : differencers) + taskExecutor.submit(differencer); + trees.clear(); // allows gc to do its thing } /** * @return true if the given node pair was the last remaining */ - synchronized boolean completedSynchronization(NodePair nodes, boolean success) + boolean completedSynchronization() { - if (!success) - failed = true; - Differencer completed = new Differencer(desc, new TreeResponse(nodes.endpoint1, null), new TreeResponse(nodes.endpoint2, null)); - differencers.remove(completed); - return differencers.size() == 0; + return waitForSync.decrementAndGet() == 0; } } diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index bd0fe3f2d7..1b6375e3cd 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -213,7 +213,7 @@ public class RepairSession extends WrappedRunnable implements IEndpointStateChan logger.debug(String.format("[repair #%s] Repair completed between %s and %s on %s", getId(), nodes.endpoint1, nodes.endpoint2, desc.columnFamily)); - if (job.completedSynchronization(nodes, success)) + if (job.completedSynchronization()) { RepairJob completedJob = syncingJobs.remove(job.desc.columnFamily); String remaining = syncingJobs.size() == 0 ? "" : String.format(" (%d remaining column family to sync for this session)", syncingJobs.size()); diff --git a/src/java/org/apache/cassandra/streaming/StreamPlan.java b/src/java/org/apache/cassandra/streaming/StreamPlan.java index ff78e84943..f9d1ae5745 100644 --- a/src/java/org/apache/cassandra/streaming/StreamPlan.java +++ b/src/java/org/apache/cassandra/streaming/StreamPlan.java @@ -34,6 +34,7 @@ public class StreamPlan { private final UUID planId = UUIDGen.getTimeUUID(); private final String description; + private final List handlers = new ArrayList<>(); // sessions per InetAddress of the other end. private final Map sessions = new HashMap<>(); @@ -130,6 +131,14 @@ public class StreamPlan return this; } + public StreamPlan listeners(StreamEventHandler handler, StreamEventHandler... handlers) + { + this.handlers.add(handler); + if (handlers != null) + Collections.addAll(this.handlers, handlers); + return this; + } + /** * @return true if this plan has no plan to execute */ @@ -145,7 +154,7 @@ public class StreamPlan */ public StreamResultFuture execute() { - return StreamResultFuture.init(planId, description, sessions.values()); + return StreamResultFuture.init(planId, description, sessions.values(), handlers); } /** diff --git a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java index ccd3c92aee..dcffaff54c 100644 --- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java +++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java @@ -75,9 +75,14 @@ public final class StreamResultFuture extends AbstractFuture set(getCurrentState()); } - static StreamResultFuture init(UUID planId, String description, Collection sessions) + static StreamResultFuture init(UUID planId, String description, Collection sessions, Collection listeners) { StreamResultFuture future = createAndRegister(planId, description, sessions); + if (listeners != null) + { + for (StreamEventHandler listener : listeners) + future.addEventListener(listener); + } logger.info("[Stream #{}] Executing streaming plan for {}", planId, description); // start sessions diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java index 623f7cced9..01952e5a80 100644 --- a/src/java/org/apache/cassandra/tools/BulkLoader.java +++ b/src/java/org/apache/cassandra/tools/BulkLoader.java @@ -78,7 +78,10 @@ public class BulkLoader StreamResultFuture future = null; try { - future = loader.stream(options.ignores); + if (options.noProgress) + future = loader.stream(options.ignores); + else + future = loader.stream(options.ignores, new ProgressIndicator()); } catch (Exception e) { @@ -93,8 +96,6 @@ public class BulkLoader } handler.output(String.format("Streaming session ID: %s", future.planId)); - if (!options.noProgress) - future.addEventListener(new ProgressIndicator()); try { diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java index 45b670d05a..22850bd054 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java @@ -19,12 +19,14 @@ package org.apache.cassandra.io.compress; import java.io.*; +import java.util.Collections; import java.util.Random; import org.junit.Test; import org.apache.cassandra.db.composites.*; import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.io.util.*; @@ -49,6 +51,46 @@ public class CompressedRandomAccessReaderTest testResetAndTruncate(File.createTempFile("compressed", "1"), true, 10); testResetAndTruncate(File.createTempFile("compressed", "2"), true, CompressionParameters.DEFAULT_CHUNK_LENGTH); } + @Test + public void test6791() throws IOException, ConfigurationException + { + File f = File.createTempFile("compressed6791_", "3"); + String filename = f.getAbsolutePath(); + try + { + + MetadataCollector sstableMetadataCollector = new MetadataCollector(new SimpleDenseCellNameType(BytesType.instance)).replayPosition(null); + CompressedSequentialWriter writer = new CompressedSequentialWriter(f, filename + ".metadata", false, new CompressionParameters(SnappyCompressor.instance, 32, Collections.emptyMap()), sstableMetadataCollector); + + for (int i = 0; i < 20; i++) + writer.write("x".getBytes()); + + FileMark mark = writer.mark(); + // write enough garbage to create new chunks: + for (int i = 0; i < 40; ++i) + writer.write("y".getBytes()); + + writer.resetAndTruncate(mark); + + for (int i = 0; i < 20; i++) + writer.write("x".getBytes()); + writer.close(); + + CompressedRandomAccessReader reader = CompressedRandomAccessReader.open(filename, new CompressionMetadata(filename + ".metadata", f.length(), true)); + String res = reader.readLine(); + assertEquals(res, "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"); + assertEquals(40, res.length()); + } + finally + { + // cleanup + if (f.exists()) + f.delete(); + File metadata = new File(filename+ ".metadata"); + if (metadata.exists()) + metadata.delete(); + } + } private void testResetAndTruncate(File f, boolean compressed, int junkSize) throws IOException {