Merge branch 'cassandra-2.0' into cassandra-2.1

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
	src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java

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

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"

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 last query should return one row (but with c == null). Adding the marker with the insert make sure
-        // the marker with the insert make sure the semantic is correct (while making sure a
+        // the semantic is correct (while making sure a 'DELETE FROM t WHERE k = 1' does remove the row entirely)
-        // '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<Operation> updates = getOperations();

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);

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<OnDiskAtom> implements OnDiskA
         }
     }
 
+    static int indexFor(SSTableReader sstable, Composite name, List<IndexHelper.IndexInfo> 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<OnDiskAtom> 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<OnDiskAtom> 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<OnDiskAtom> 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;
                 }
             }

src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java

@@ -186,13 +186,15 @@ public class SSTableNamesIterator extends AbstractIterator<OnDiskAtom> 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;
         }

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);
     }
 
     /**

src/java/org/apache/cassandra/io/sstable/SSTableLoader.java

@@ -145,7 +145,7 @@ public class SSTableLoader implements StreamEventHandler
         return stream(Collections.<InetAddress>emptySet());
     }
 
-    public StreamResultFuture stream(Set<InetAddress> toIgnore)
+    public StreamResultFuture stream(Set<InetAddress> 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();
+        plan.listeners(this, listeners);
-        bulkResult.addEventListener(this);
+        return plan.execute();
-        return bulkResult;
     }
 
     public void onSuccess(StreamState finalState) {}

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<TreeResponse> 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<Differencer> 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<Differencer> 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)
+        return waitForSync.decrementAndGet() == 0;
-            failed = true;
-        Differencer completed = new Differencer(desc, new TreeResponse(nodes.endpoint1, null), new TreeResponse(nodes.endpoint2, null));
-        differencers.remove(completed);
-        return differencers.size() == 0;
     }
 }

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());

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<StreamEventHandler> handlers = new ArrayList<>();
 
     // sessions per InetAddress of the other end.
     private final Map<InetAddress, StreamSession> 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);
     }
 
     /**

src/java/org/apache/cassandra/streaming/StreamResultFuture.java

@@ -75,9 +75,14 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
             set(getCurrentState());
     }
 
-    static StreamResultFuture init(UUID planId, String description, Collection<StreamSession> sessions)
+    static StreamResultFuture init(UUID planId, String description, Collection<StreamSession> sessions, Collection<StreamEventHandler> 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

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
         {

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.<String, String>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
     {