diff --git a/CHANGES.txt b/CHANGES.txt index 9c626798b2..00848f6467 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -49,6 +49,8 @@ Merged from 1.0: * ensure that directory is selected for compaction for user-defined tasks and upgradesstables (CASSANDRA-3985) * fix NPE on invalid CQL delete command (CASSANDRA-3755) + * allow custom types in CLI's assume command (CASSANDRA-4081) + * Fix totalBytes count for parallel compactions (CASSANDRA-3758) 1.1-beta1 diff --git a/src/java/org/apache/cassandra/cli/Cli.g b/src/java/org/apache/cassandra/cli/Cli.g index e7cba6c04c..742ccf2e1b 100644 --- a/src/java/org/apache/cassandra/cli/Cli.g +++ b/src/java/org/apache/cassandra/cli/Cli.g @@ -301,8 +301,8 @@ truncateStatement ; assumeStatement - : ASSUME columnFamily assumptionElement=Identifier 'AS' defaultType=Identifier - -> ^(NODE_ASSUME columnFamily $assumptionElement $defaultType) + : ASSUME columnFamily assumptionElement=Identifier 'AS' entityName + -> ^(NODE_ASSUME columnFamily $assumptionElement entityName) ; consistencyLevelStatement diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java index ba772a9e8e..4e97e82826 100644 --- a/src/java/org/apache/cassandra/cli/CliClient.java +++ b/src/java/org/apache/cassandra/cli/CliClient.java @@ -1474,17 +1474,25 @@ public class CliClient AbstractType comparator; // Could be UTF8Type, IntegerType, LexicalUUIDType etc. - String defaultType = statement.getChild(2).getText(); + String defaultType = CliUtils.unescapeSQLString(statement.getChild(2).getText()); try { - comparator = Function.valueOf(defaultType.toUpperCase()).getValidator(); + comparator = TypeParser.parse(defaultType); } - catch (Exception e) + catch (ConfigurationException e) { - String functions = Function.getFunctionNames(); - sessionState.out.println("Type '" + defaultType + "' was not found. Available: " + functions); - return; + try + { + comparator = Function.valueOf(defaultType.toUpperCase()).getValidator(); + } + catch (Exception ne) + { + String functions = Function.getFunctionNames(); + sessionState.out.println("Type '" + defaultType + "' was not found. Available: " + functions + + " Or any class which extends o.a.c.db.marshal.AbstractType."); + return; + } } // making string representation look property e.g. o.a.c.db.marshal.UTF8Type diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionIterable.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionIterable.java index 5e7b0888e9..e05a64c323 100644 --- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionIterable.java +++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionIterable.java @@ -41,15 +41,24 @@ public abstract class AbstractCompactionIterable extends CompactionInfo.Holder i protected final OperationType type; protected final CompactionController controller; - protected long totalBytes; + protected final long totalBytes; protected volatile long bytesRead; + protected final List scanners; protected final Throttle throttle; - public AbstractCompactionIterable(CompactionController controller, OperationType type) + public AbstractCompactionIterable(CompactionController controller, OperationType type, List scanners) { this.controller = controller; this.type = type; + this.scanners = scanners; + this.bytesRead = 0; + + long bytes = 0; + for (SSTableScanner scanner : scanners) + bytes += scanner.getFileLength(); + this.totalBytes = bytes; + this.throttle = new Throttle(toString(), new Throttle.ThroughputFunction() { /** @return Instantaneous throughput target in bytes per millisecond. */ diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterable.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterable.java index 2fd0240535..eb88489cb2 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterable.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterable.java @@ -41,7 +41,6 @@ public class CompactionIterable extends AbstractCompactionIterable private static Logger logger = LoggerFactory.getLogger(CompactionIterable.class); private long row; - private final List scanners; private static final Comparator comparator = new Comparator() { @@ -58,12 +57,8 @@ public class CompactionIterable extends AbstractCompactionIterable protected CompactionIterable(OperationType type, List scanners, CompactionController controller) { - super(controller, type); - this.scanners = scanners; + super(controller, type, scanners); row = 0; - totalBytes = bytesRead = 0; - for (SSTableScanner scanner : scanners) - totalBytes += scanner.getFileLength(); } protected static List getScanners(Iterable sstables) throws IOException diff --git a/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java index 9b67676591..ba85132e34 100644 --- a/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java +++ b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java @@ -59,7 +59,6 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable { private static Logger logger = LoggerFactory.getLogger(ParallelCompactionIterable.class); - private final List scanners; private final int maxInMemorySize; public ParallelCompactionIterable(OperationType type, Iterable sstables, CompactionController controller) throws IOException @@ -74,8 +73,7 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable protected ParallelCompactionIterable(OperationType type, List scanners, CompactionController controller, int maxInMemorySize) { - super(controller, type); - this.scanners = scanners; + super(controller, type, scanners); this.maxInMemorySize = maxInMemorySize; } diff --git a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java index edae8767d3..f329636db6 100644 --- a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java +++ b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java @@ -46,7 +46,6 @@ public class IncomingTcpConnection extends Thread { assert socket != null; this.socket = socket; - from = socket.getInetAddress(); // maximize chance of this not being nulled by disconnect } /** @@ -92,6 +91,7 @@ public class IncomingTcpConnection extends Thread input = new DataInputStream(new BufferedInputStream(socket.getInputStream(), 4096)); // Receive the first message to set the version. Message msg = receiveMessage(input, version); + from = msg.getFrom(); // why? see => CASSANDRA-4099 if (version > MessagingService.version_) { // save the endpoint so gossip will reconnect to it @@ -100,7 +100,7 @@ public class IncomingTcpConnection extends Thread } else if (msg != null) { - Gossiper.instance.setVersion(msg.getFrom(), version); + Gossiper.instance.setVersion(from, version); logger.debug("set version for {} to {}", from, version); }