cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From marc...@apache.org
Subject [14/15] cassandra git commit: Merge branch 'cassandra-3.0' into cassandra-3.5
Date Tue, 29 Mar 2016 09:13:00 GMT
Merge branch 'cassandra-3.0' into cassandra-3.5


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/0a250854
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/0a250854
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/0a250854

Branch: refs/heads/trunk
Commit: 0a2508544069f035dc43a81b8700c0b7bf2d28e3
Parents: c7ef7c9 0c91977
Author: Marcus Eriksson <marcuse@apache.org>
Authored: Tue Mar 29 11:05:46 2016 +0200
Committer: Marcus Eriksson <marcuse@apache.org>
Committed: Tue Mar 29 11:05:46 2016 +0200

----------------------------------------------------------------------
 CHANGES.txt                                     |  2 +
 .../apache/cassandra/db/ColumnFamilyStore.java  | 20 ++++-----
 .../db/compaction/CompactionManager.java        | 44 +++++++++++++-------
 .../cassandra/service/StorageService.java       | 30 ++++++++++---
 .../cassandra/service/StorageServiceMBean.java  |  8 ++++
 .../org/apache/cassandra/tools/NodeProbe.java   | 38 ++++++++++-------
 .../cassandra/tools/nodetool/Cleanup.java       |  9 +++-
 .../tools/nodetool/RelocateSSTables.java        |  8 +++-
 .../apache/cassandra/tools/nodetool/Scrub.java  | 13 ++++--
 .../tools/nodetool/UpgradeSSTable.java          | 10 ++++-
 .../org/apache/cassandra/db/CleanupTest.java    |  6 +--
 .../unit/org/apache/cassandra/db/ScrubTest.java | 16 +++----
 12 files changed, 141 insertions(+), 63 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 2907df9,fdc873e..3e69959
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -25,43 -23,14 +25,45 @@@ Merged from 2.2
   * Only log yaml config once, at startup (CASSANDRA-11217)
   * Reference leak with parallel repairs on the same table (CASSANDRA-11215)
  Merged from 2.1:
+  * Add a -j parameter to scrub/cleanup/upgradesstables to state how
+    many threads to use (CASSANDRA-11179)
 - * Backport CASSANDRA-10679 (CASSANDRA-9598)
 - * InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
   * COPY FROM on large datasets: fix progress report and debug performance (CASSANDRA-11053)
 + * InvalidateKeys should have a weak ref to key cache (CASSANDRA-11176)
  
 -3.0.4
 - * Preserve order for preferred SSL cipher suites (CASSANDRA-11164)
 +
 +3.4
 + * (cqlsh) add cqlshrc option to always connect using ssl (CASSANDRA-10458)
 + * Cleanup a few resource warnings (CASSANDRA-11085)
 + * Allow custom tracing implementations (CASSANDRA-10392)
 + * Extract LoaderOptions to be able to be used from outside (CASSANDRA-10637)
 + * fix OnDiskIndexTest to properly treat empty ranges (CASSANDRA-11205)
 + * fix TrackerTest to handle new notifications (CASSANDRA-11178)
 + * add SASI validation for partitioner and complex columns (CASSANDRA-11169)
 + * Add caching of encrypted credentials in PasswordAuthenticator (CASSANDRA-7715)
 + * fix SASI memtable switching on flush (CASSANDRA-11159)
 + * Remove duplicate offline compaction tracking (CASSANDRA-11148)
 + * fix EQ semantics of analyzed SASI indexes (CASSANDRA-11130)
 + * Support long name output for nodetool commands (CASSANDRA-7950)
 + * Encrypted hints (CASSANDRA-11040)
 + * SASI index options validation (CASSANDRA-11136)
 + * Optimize disk seek using min/max column name meta data when the LIMIT clause is used
 +   (CASSANDRA-8180)
 + * Add LIKE support to CQL3 (CASSANDRA-11067)
 + * Generic Java UDF types (CASSANDRA-10819)
 + * cqlsh: Include sub-second precision in timestamps by default (CASSANDRA-10428)
 + * Set javac encoding to utf-8 (CASSANDRA-11077)
 + * Integrate SASI index into Cassandra (CASSANDRA-10661)
 + * Add --skip-flush option to nodetool snapshot
 + * Skip values for non-queried columns (CASSANDRA-10657)
 + * Add support for secondary indexes on static columns (CASSANDRA-8103)
 + * CommitLogUpgradeTestMaker creates broken commit logs (CASSANDRA-11051)
 + * Add metric for number of dropped mutations (CASSANDRA-10866)
 + * Simplify row cache invalidation code (CASSANDRA-10396)
 + * Support user-defined compaction through nodetool (CASSANDRA-10660)
 + * Stripe view locks by key and table ID to reduce contention (CASSANDRA-10981)
 + * Add nodetool gettimeout and settimeout commands (CASSANDRA-10953)
 + * Add 3.0 metadata to sstablemetadata output (CASSANDRA-10838)
 +Merged from 3.0:
   * MV should only query complex columns included in the view (CASSANDRA-11069)
   * Failed aggregate creation breaks server permanently (CASSANDRA-11064)
   * Add sstabledump tool (CASSANDRA-7464)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index fbfb7ee,a1db5b5..f193c4d
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@@ -1436,16 -1363,11 +1436,16 @@@ public class ColumnFamilyStore implemen
          return CompactionManager.instance.performVerify(ColumnFamilyStore.this, extendedVerify);
      }
  
-     public CompactionManager.AllSSTableOpStatus sstablesRewrite(boolean excludeCurrentVersion)
throws ExecutionException, InterruptedException
+     public CompactionManager.AllSSTableOpStatus sstablesRewrite(boolean excludeCurrentVersion,
int jobs) throws ExecutionException, InterruptedException
      {
-         return CompactionManager.instance.performSSTableRewrite(ColumnFamilyStore.this,
excludeCurrentVersion);
+         return CompactionManager.instance.performSSTableRewrite(ColumnFamilyStore.this,
excludeCurrentVersion, jobs);
      }
  
-     public CompactionManager.AllSSTableOpStatus relocateSSTables() throws ExecutionException,
InterruptedException
++    public CompactionManager.AllSSTableOpStatus relocateSSTables(int jobs) throws ExecutionException,
InterruptedException
 +    {
-         return CompactionManager.instance.relocateSSTables(this);
++        return CompactionManager.instance.relocateSSTables(this, jobs);
 +    }
 +
      public void markObsolete(Collection<SSTableReader> sstables, OperationType compactionType)
      {
          assert !sstables.isEmpty();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 627ae32,f339660..f9b0997
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@@ -438,80 -451,9 +452,80 @@@ public class CompactionManager implemen
                  CleanupStrategy cleanupStrategy = CleanupStrategy.get(cfStore, ranges, FBUtilities.nowInSeconds());
                  doCleanupOne(cfStore, txn, cleanupStrategy, ranges, hasIndexes);
              }
-         }, OperationType.CLEANUP);
+         }, jobs, OperationType.CLEANUP);
      }
  
-     public AllSSTableOpStatus relocateSSTables(final ColumnFamilyStore cfs) throws ExecutionException,
InterruptedException
++    public AllSSTableOpStatus relocateSSTables(final ColumnFamilyStore cfs, int jobs) throws
ExecutionException, InterruptedException
 +    {
 +        if (!cfs.getPartitioner().splitter().isPresent())
 +        {
 +            logger.info("Partitioner does not support splitting");
 +            return AllSSTableOpStatus.ABORTED;
 +        }
 +        final Collection<Range<Token>> r = StorageService.instance.getLocalRanges(cfs.keyspace.getName());
 +
 +        if (r.isEmpty())
 +        {
 +            logger.info("Relocate cannot run before a node has joined the ring");
 +            return AllSSTableOpStatus.ABORTED;
 +        }
 +
 +        final List<Range<Token>> localRanges = Range.sort(r);
 +        final Directories.DataDirectory[] locations = cfs.getDirectories().getWriteableLocations();
 +        final List<PartitionPosition> diskBoundaries = StorageService.getDiskBoundaries(localRanges,
cfs.getPartitioner(), locations);
 +
 +        return parallelAllSSTableOperation(cfs, new OneSSTableOperation()
 +        {
 +            @Override
 +            public Iterable<SSTableReader> filterSSTables(LifecycleTransaction transaction)
 +            {
 +                Set<SSTableReader> originals = Sets.newHashSet(transaction.originals());
 +                Set<SSTableReader> needsRelocation = originals.stream().filter(s ->
!inCorrectLocation(s)).collect(Collectors.toSet());
 +                transaction.cancel(Sets.difference(originals, needsRelocation));
 +
 +                Map<Integer, List<SSTableReader>> groupedByDisk = needsRelocation.stream().collect(Collectors.groupingBy((s)
->
 +                        CompactionStrategyManager.getCompactionStrategyIndex(cfs, cfs.getDirectories(),
s)));
 +
 +                int maxSize = 0;
 +                for (List<SSTableReader> diskSSTables : groupedByDisk.values())
 +                    maxSize = Math.max(maxSize, diskSSTables.size());
 +
 +                List<SSTableReader> mixedSSTables = new ArrayList<>();
 +
 +                for (int i = 0; i < maxSize; i++)
 +                    for (List<SSTableReader> diskSSTables : groupedByDisk.values())
 +                        if (i < diskSSTables.size())
 +                            mixedSSTables.add(diskSSTables.get(i));
 +
 +                return mixedSSTables;
 +            }
 +
 +            private boolean inCorrectLocation(SSTableReader sstable)
 +            {
 +                if (!cfs.getPartitioner().splitter().isPresent())
 +                    return true;
 +                int directoryIndex = CompactionStrategyManager.getCompactionStrategyIndex(cfs,
cfs.getDirectories(), sstable);
 +                Directories.DataDirectory[] locations = cfs.getDirectories().getWriteableLocations();
 +
 +                Directories.DataDirectory location = locations[directoryIndex];
 +                PartitionPosition diskLast = diskBoundaries.get(directoryIndex);
 +                // the location we get from directoryIndex is based on the first key in
the sstable
 +                // now we need to make sure the last key is less than the boundary as well:
 +                return sstable.descriptor.directory.getAbsolutePath().startsWith(location.location.getAbsolutePath())
&& sstable.last.compareTo(diskLast) <= 0;
 +            }
 +
 +            @Override
 +            public void execute(LifecycleTransaction txn) throws IOException
 +            {
 +                logger.debug("Relocating {}", txn.originals());
 +                AbstractCompactionTask task = cfs.getCompactionStrategyManager().getCompactionTask(txn,
NO_GC, Long.MAX_VALUE);
 +                task.setUserDefined(true);
 +                task.setCompactionType(OperationType.RELOCATE);
 +                task.execute(metrics);
 +            }
-         }, OperationType.RELOCATE);
++        }, jobs, OperationType.RELOCATE);
 +    }
 +
      public ListenableFuture<?> submitAntiCompaction(final ColumnFamilyStore cfs,
                                            final Collection<Range<Token>> ranges,
                                            final Refs<SSTableReader> sstables,

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageService.java
index 22d67e5,75573ac..52dcb97
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@@ -2789,59 -2712,6 +2804,64 @@@ public class StorageService extends Not
          }
      }
  
 +    public int relocateSSTables(String keyspaceName, String ... columnFamilies) throws IOException,
ExecutionException, InterruptedException
 +    {
++        return relocateSSTables(0, keyspaceName, columnFamilies);
++    }
++
++    public int relocateSSTables(int jobs, String keyspaceName, String ... columnFamilies)
throws IOException, ExecutionException, InterruptedException
++    {
 +        CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL;
 +        for (ColumnFamilyStore cfs : getValidColumnFamilies(false, false, keyspaceName,
columnFamilies))
 +        {
-             CompactionManager.AllSSTableOpStatus oneStatus = cfs.relocateSSTables();
++            CompactionManager.AllSSTableOpStatus oneStatus = cfs.relocateSSTables(jobs);
 +            if (oneStatus != CompactionManager.AllSSTableOpStatus.SUCCESSFUL)
 +                status = oneStatus;
 +        }
 +        return status.statusCode;
 +    }
 +
 +    /**
 +     * Takes the snapshot of a multiple column family from different keyspaces. A snapshot
name must be specified.
 +     *
 +     * @param tag
 +     *            the tag given to the snapshot; may not be null or empty
 +     * @param options
 +     *            Map of options (skipFlush is the only supported option for now)
 +     * @param entities
 +     *            list of keyspaces / tables in the form of empty | ks1 ks2 ... | ks1.cf1,ks2.cf2,...
 +     */
 +    @Override
 +    public void takeSnapshot(String tag, Map<String, String> options, String... entities)
throws IOException
 +    {
 +        boolean skipFlush = Boolean.parseBoolean(options.getOrDefault("skipFlush", "false"));
 +
 +        if (entities != null && entities.length > 0 && entities[0].contains("."))
 +        {
 +            takeMultipleTableSnapshot(tag, skipFlush, entities);
 +        }
 +        else
 +        {
 +            takeSnapshot(tag, skipFlush, entities);
 +        }
 +    }
 +
 +    /**
 +     * Takes the snapshot of a specific table. A snapshot name must be
 +     * specified.
 +     *
 +     * @param keyspaceName
 +     *            the keyspace which holds the specified table
 +     * @param tableName
 +     *            the table to snapshot
 +     * @param tag
 +     *            the tag given to the snapshot; may not be null or empty
 +     */
 +    public void takeTableSnapshot(String keyspaceName, String tableName, String tag)
 +            throws IOException {
 +        takeMultipleTableSnapshot(tag, false, keyspaceName + "." + tableName);
 +    }
 +
      /**
       * Takes the snapshot for the given keyspaces. A snapshot name must be specified.
       *

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/src/java/org/apache/cassandra/service/StorageServiceMBean.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/service/StorageServiceMBean.java
index dc12253,5b7331d..2238f33
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@@ -251,7 -248,6 +251,9 @@@ public interface StorageServiceMBean ex
       */
      public void forceKeyspaceCompaction(boolean splitOutput, String keyspaceName, String...
tableNames) throws IOException, ExecutionException, InterruptedException;
  
++    @Deprecated
 +    public int relocateSSTables(String keyspace, String ... cfnames) throws IOException,
ExecutionException, InterruptedException;
++    public int relocateSSTables(int jobs, String keyspace, String ... cfnames) throws IOException,
ExecutionException, InterruptedException;
      /**
       * Trigger a cleanup of keys on a single keyspace
       */

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/src/java/org/apache/cassandra/tools/NodeProbe.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tools/NodeProbe.java
index 405f70c,a9d71d8..7d15926
--- a/src/java/org/apache/cassandra/tools/NodeProbe.java
+++ b/src/java/org/apache/cassandra/tools/NodeProbe.java
@@@ -304,11 -309,6 +314,11 @@@ public class NodeProbe implements AutoC
          ssProxy.forceKeyspaceCompaction(splitOutput, keyspaceName, tableNames);
      }
  
-     public void relocateSSTables(String keyspace, String[] cfnames) throws IOException,
ExecutionException, InterruptedException
++    public void relocateSSTables(int jobs, String keyspace, String[] cfnames) throws IOException,
ExecutionException, InterruptedException
 +    {
-         ssProxy.relocateSSTables(keyspace, cfnames);
++        ssProxy.relocateSSTables(jobs, keyspace, cfnames);
 +    }
 +
      public void forceKeyspaceFlush(String keyspaceName, String... tableNames) throws IOException,
ExecutionException, InterruptedException
      {
          ssProxy.forceKeyspaceFlush(keyspaceName, tableNames);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/src/java/org/apache/cassandra/tools/nodetool/RelocateSSTables.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/tools/nodetool/RelocateSSTables.java
index 8522bc4,0000000..7c3066c
mode 100644,000000..100644
--- a/src/java/org/apache/cassandra/tools/nodetool/RelocateSSTables.java
+++ b/src/java/org/apache/cassandra/tools/nodetool/RelocateSSTables.java
@@@ -1,49 -1,0 +1,55 @@@
 +/*
 + * 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.tools.nodetool;
 +
 +import java.util.ArrayList;
 +import java.util.List;
 +
 +import io.airlift.command.Arguments;
 +import io.airlift.command.Command;
++import io.airlift.command.Option;
 +import org.apache.cassandra.tools.NodeProbe;
 +import org.apache.cassandra.tools.NodeTool;
 +
 +@Command(name = "relocatesstables", description = "Relocates sstables to the correct disk")
 +public class RelocateSSTables extends NodeTool.NodeToolCmd
 +{
 +    @Arguments(usage = "<keyspace> <table>", description = "The keyspace and
table name")
 +    private List<String> args = new ArrayList<>();
 +
++    @Option(title = "jobs",
++            name = {"-j", "--jobs"},
++            description = "Number of sstables to relocate simultanously, set to 0 to use
all available compaction threads")
++    private int jobs = 2;
++
 +    @Override
 +    public void execute(NodeProbe probe)
 +    {
 +        List<String> keyspaces = parseOptionalKeyspace(args, probe);
 +        String[] cfnames = parseOptionalTables(args);
 +        try
 +        {
 +            for (String keyspace : keyspaces)
-                 probe.relocateSSTables(keyspace, cfnames);
++                probe.relocateSSTables(jobs, keyspace, cfnames);
 +        }
 +        catch (Exception e)
 +        {
 +            throw new RuntimeException("Got error while relocating", e);
 +        }
 +    }
 +}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/test/unit/org/apache/cassandra/db/CleanupTest.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/0a250854/test/unit/org/apache/cassandra/db/ScrubTest.java
----------------------------------------------------------------------


Mime
View raw message