diff --git a/CHANGES.txt b/CHANGES.txt index 5456705f47..5bf6bf5793 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,11 +1,16 @@ 0.8.1 - * add support for insert, delete in cql BATCH (CASSANDRA-2537) - * add support for IN to cql SELECT, UPDATE (CASSANDRA-2553) - * add timestamp support to cql INSERT, UPDATE, and BATCH (CASSANDRA-2555) + * CQL: + - support for insert, delete in BATCH (CASSANDRA-2537) + - support for IN to SELECT, UPDATE (CASSANDRA-2553) + - timestamp support for INSERT, UPDATE, and BATCH (CASSANDRA-2555) + - TTL support (CASSANDRA-2476) + - counter support (CASSANDRA-2473) + - improve JDBC spec compliance (CASSANDRA-2720) + - ALTER TABLE (CASSANDRA-1709) + - DROP INDEX (CASSANDRA-2617) * add support for comparator parameters and a generic ReverseType (CASSANDRA-2355) * add CompositeType and DynamicCompositeType (CASSANDRA-2231) - * add CQL TTL support (CASSANDRA-2476) * optimize batches containing multiple updates to the same row (CASSANDRA-2583) * adjust hinted handoff page size to avoid OOM with large columns @@ -31,6 +36,9 @@ * close scrub file handles (CASSANDRA-2669) * throttle migration replay (CASSANDRA-2714) * optimize column serializer creation (CASSANDRA-2716) + * fix truncate/compaction race (CASSANDRA-2673) + * workaround large resultsets causing large allocation retention + by nio sockets (CASSANDRA-2654) 0.8.0-final @@ -42,10 +50,11 @@ * switch to native Thrift for Hadoop map/reduce (CASSANDRA-2667) * fix StackOverflowError when building from eclipse (CASSANDRA-2687) * only provide replication_factor to strategy_options "help" for - SimpleStrategy, OldNetworkTopologyStrategy (CASSANDRA-2678) + SimpleStrategy, OldNetworkTopologyStrategy (CASSANDRA-2678, 2713) * fix exception adding validators to non-string columns (CASSANDRA-2696) * avoid instantiating DatabaseDescriptor in JDBC (CASSANDRA-2694) * fix potential stack overflow during compaction (CASSANDRA-2626) + * clone super columns to avoid modifying them during flush (CASSANDRA-2675) * reset underlying iterator in EchoedRow constructor (CASSANDRA-2653) @@ -147,6 +156,9 @@ * reduce contention on Table.flusherLock (CASSANDRA-1954) * try harder to detect failures during streaming, cleaning up temporary files more reliably (CASSANDRA-2088) + + +0.6.13 * shut down server for OOM on a Thrift thread (CASSANDRA-2269) * fix tombstone handling in repair and sstable2json (CASSANDRA-2279) * preserve version when streaming data from old sstables (CASSANDRA-2283) diff --git a/build.xml b/build.xml index 47c0dd4c3a..173dd043c4 100644 --- a/build.xml +++ b/build.xml @@ -64,7 +64,7 @@ - + @@ -693,7 +693,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} destdir="${build.classes.main}" includeantruntime="false"> - - + + + + @@ -1024,6 +1026,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} + diff --git a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 76ec3b9158..6bba368ed4 100644 --- a/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -357,12 +357,12 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface Mutation mutation = new Mutation(); if (DataType.findType(pair.get(1)) == DataType.BAG) // supercolumn { - org.apache.cassandra.hadoop.avro.SuperColumn sc = new org.apache.cassandra.hadoop.avro.SuperColumn(); + org.apache.cassandra.thrift.SuperColumn sc = new org.apache.cassandra.thrift.SuperColumn(); sc.name = objToBB(pair.get(0)); - ArrayList columns = new ArrayList(); + ArrayList columns = new ArrayList(); for (Tuple subcol : (DefaultDataBag) pair.get(1)) { - org.apache.cassandra.hadoop.avro.Column column = new org.apache.cassandra.hadoop.avro.Column(); + org.apache.cassandra.thrift.Column column = new org.apache.cassandra.thrift.Column(); column.name = objToBB(subcol.get(0)); column.value = objToBB(subcol.get(1)); column.setTimestamp(System.currentTimeMillis() * 1000); diff --git a/debian/changelog b/debian/changelog index aa2fba7304..fd4892a03c 100644 --- a/debian/changelog +++ b/debian/changelog @@ -2,7 +2,7 @@ cassandra (0.8.0) unstable; urgency=low * New release - -- Eric Evans Mon, 23 May 2011 15:59:48 -0500 + -- Eric Evans Mon, 30 May 2011 12:58:23 -0500 cassandra (0.8.0~rc1) unstable; urgency=low diff --git a/doc/cql/CQL.textile b/doc/cql/CQL.textile index 6f33d99e26..63a72900f7 100644 --- a/doc/cql/CQL.textile +++ b/doc/cql/CQL.textile @@ -69,6 +69,17 @@ SELECT ... WHERE [LIMIT N] ... Limiting the number of rows returned can be achieved by adding the @LIMIT@ option to a @SELECT@ expression. @LIMIT@ defaults to 10,000 when left unset. +h2. ALTER TABLE + +_Synopsis:_ + +bc. +ALTER TABLE ADD ; +ALTER TABLE ALTER TYPE ; +ALTER TABLE DROP ; + +An @ALTER@ is used to manipulate with ColumnFamily columns. It allows you to add new columns, alter and drop existing columns. No results are returned. + h2. INSERT _Synopsis:_ @@ -282,6 +293,15 @@ bc. CREATE INDEX [index_name] ON (column_name); A @CREATE INDEX@ statement is used to create a new, automatic secondary index for the named column. +h2. DROP INDEX + +_Synopsis:_ + +bc. DROP INDEX + +A @DROP INDEX@ statement is used to drop an existing secondary index. +DROP INDEX statement will search all ColumnFamilies in the current Keyspace for specified index and delete it if found. + h2. DROP _Synopsis:_ diff --git a/drivers/java/src/org/apache/cassandra/cql/jdbc/CResultSet.java b/drivers/java/src/org/apache/cassandra/cql/jdbc/CResultSet.java index 6c939c2819..0b999faae1 100644 --- a/drivers/java/src/org/apache/cassandra/cql/jdbc/CResultSet.java +++ b/drivers/java/src/org/apache/cassandra/cql/jdbc/CResultSet.java @@ -48,6 +48,7 @@ public class CResultSet extends AbstractResultSet implements CassandraResultSet // the current row key when iterating through results. private byte[] curRowKey = null; + private TypedColumn typedCurRowKey = null; /** The values. */ private List values = new ArrayList(); @@ -78,6 +79,11 @@ public class CResultSet extends AbstractResultSet implements CassandraResultSet return curRowKey; } + public TypedColumn getTypedKey() + { + return typedCurRowKey; + } + public TypedColumn getColumn(int i) { return values.get(i); @@ -379,6 +385,8 @@ public class CResultSet extends AbstractResultSet implements CassandraResultSet CqlRow row = rSetIter.next(); rowNumber++; curRowKey = row.getKey(); + typedCurRowKey = decoder.makeKeyColumn(keyspace, columnFamily, curRowKey); + List cols = row.getColumns(); for (Column col : cols) { diff --git a/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraConnection.java b/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraConnection.java index b28c298a7a..7a5baa5df9 100644 --- a/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraConnection.java +++ b/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraConnection.java @@ -28,8 +28,10 @@ import java.sql.Connection; import java.sql.DatabaseMetaData; import java.sql.NClob; import java.sql.PreparedStatement; +import java.sql.ResultSet; import java.sql.SQLClientInfoException; import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; import java.sql.SQLWarning; import java.sql.SQLXML; import java.sql.Savepoint; @@ -52,10 +54,13 @@ import org.apache.thrift.transport.TTransportException; */ class CassandraConnection implements Connection { - - /** The cassandra con. */ + private Properties clientInfo = new Properties(); + + /** + * The cassandra con. + */ private org.apache.cassandra.cql.jdbc.Connection cassandraCon; - + /** * Instantiates a new cassandra connection. * @@ -77,8 +82,7 @@ class CassandraConnection implements Connection final int host_backwardIdx = host_port.indexOf('/'); final String port = host_port.substring(host_colonIdx + 1, host_backwardIdx); final String keyspace = host_port.substring(host_backwardIdx + 1); - cassandraCon = new org.apache.cassandra.cql.jdbc.Connection(hostName, Integer.valueOf(port), userName, - password); + cassandraCon = new org.apache.cassandra.cql.jdbc.Connection(hostName, Integer.valueOf(port), userName, password); final String useQ = "USE " + keyspace; cassandraCon.execute(useQ); } @@ -120,7 +124,7 @@ class CassandraConnection implements Connection } } - + /** * @param arg0 * @return @@ -128,10 +132,10 @@ class CassandraConnection implements Connection */ public boolean isWrapperFor(Class arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLException("no object is found that implements this interface"); } - + /** * @param * @param arg0 @@ -140,17 +144,19 @@ class CassandraConnection implements Connection */ public T unwrap(Class arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLException("no object is found that implements this interface"); } - /** * @throws SQLException */ public void clearWarnings() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + // This implementation does not support the collection of warnings so clearing is a no-op + // but it is still an exception to call this on a closed connection. + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); } /** @@ -172,7 +178,7 @@ class CassandraConnection implements Connection */ public void commit() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLException("The Cassandra Implementation is always in auto-commit mode."); } @@ -184,7 +190,7 @@ class CassandraConnection implements Connection */ public Array createArrayOf(String arg0, Object[] arg1) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -194,7 +200,7 @@ class CassandraConnection implements Connection */ public Blob createBlob() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -204,7 +210,7 @@ class CassandraConnection implements Connection */ public Clob createClob() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -214,7 +220,7 @@ class CassandraConnection implements Connection */ public NClob createNClob() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -224,10 +230,10 @@ class CassandraConnection implements Connection */ public SQLXML createSQLXML() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } - + /** * @return * @throws SQLException @@ -246,7 +252,7 @@ class CassandraConnection implements Connection */ public Statement createStatement(int arg0, int arg1) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -259,7 +265,7 @@ class CassandraConnection implements Connection */ public Statement createStatement(int arg0, int arg1, int arg2) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -271,7 +277,7 @@ class CassandraConnection implements Connection */ public Struct createStruct(String arg0, Object[] arg1) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -281,7 +287,7 @@ class CassandraConnection implements Connection */ public boolean getAutoCommit() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + return true; } @@ -291,7 +297,12 @@ class CassandraConnection implements Connection */ public String getCatalog() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + // This implementation does not support the catalog names so null is always returned if the connection is open. + // but it is still an exception to call this on a closed connection. + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + return null; } @@ -301,18 +312,24 @@ class CassandraConnection implements Connection */ public Properties getClientInfo() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + return clientInfo; } /** - * @param arg0 + * @param label * @return * @throws SQLException */ - public String getClientInfo(String arg0) throws SQLException + public String getClientInfo(String label) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + return clientInfo.getProperty(label); } @@ -322,7 +339,11 @@ class CassandraConnection implements Connection */ public int getHoldability() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is there are really no commits in Cassandra so no boundary... + return ResultSet.HOLD_CURSORS_OVER_COMMIT; } @@ -332,7 +353,11 @@ class CassandraConnection implements Connection */ public DatabaseMetaData getMetaData() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is there is no DatabaseMetaData to return but if there was we would return it... + return null; } @@ -342,7 +367,10 @@ class CassandraConnection implements Connection */ public int getTransactionIsolation() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + return Connection.TRANSACTION_NONE; } @@ -352,7 +380,7 @@ class CassandraConnection implements Connection */ public Map> getTypeMap() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -362,7 +390,11 @@ class CassandraConnection implements Connection */ public SQLWarning getWarnings() throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is there are no warnings to return in this implementation... + return null; } @@ -372,7 +404,10 @@ class CassandraConnection implements Connection */ public boolean isClosed() throws SQLException { - return false; + if (cassandraCon == null) + return true; + + return !cassandraCon.isOpen(); } @@ -387,24 +422,34 @@ class CassandraConnection implements Connection /** - * @param arg0 + * @param timeout * @return * @throws SQLException */ - public boolean isValid(int arg0) throws SQLException + public boolean isValid(int timeout) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + + if (timeout < 0) + throw new SQLException("the timeout value was less than zero"); + + // this needs to be more robust. Some query needs to be made to verify connection is really up. + return !isClosed(); } /** - * @param arg0 + * @param sql * @return * @throws SQLException */ - public String nativeSQL(String arg0) throws SQLException + public String nativeSQL(String sql) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is there are no distinction between grammars in this implementation... + // so we are just return the input argument + return sql; } @@ -415,7 +460,7 @@ class CassandraConnection implements Connection */ public CallableStatement prepareCall(String arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -428,7 +473,7 @@ class CassandraConnection implements Connection */ public CallableStatement prepareCall(String arg0, int arg1, int arg2) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -442,7 +487,7 @@ class CassandraConnection implements Connection */ public CallableStatement prepareCall(String arg0, int arg1, int arg2, int arg3) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -465,7 +510,7 @@ class CassandraConnection implements Connection */ public PreparedStatement prepareStatement(String arg0, int arg1) throws SQLException { - return null; + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -477,7 +522,7 @@ class CassandraConnection implements Connection */ public PreparedStatement prepareStatement(String arg0, int[] arg1) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -489,7 +534,7 @@ class CassandraConnection implements Connection */ public PreparedStatement prepareStatement(String arg0, String[] arg1) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -502,7 +547,7 @@ class CassandraConnection implements Connection */ public PreparedStatement prepareStatement(String arg0, int arg1, int arg2) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -516,7 +561,7 @@ class CassandraConnection implements Connection */ public PreparedStatement prepareStatement(String arg0, int arg1, int arg2, int arg3) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } @@ -526,18 +571,16 @@ class CassandraConnection implements Connection */ public void releaseSavepoint(Savepoint arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); - + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } /** * @throws SQLException - */ + */ public void rollback() throws SQLException { - throw new UnsupportedOperationException("method not supported"); - + throw new SQLException("the Cassandra Implementation is always in auto-commit mode"); } @@ -547,19 +590,18 @@ class CassandraConnection implements Connection */ public void rollback(Savepoint arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); - + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } /** - * @param arg0 + * @param autoCommit * @throws SQLException */ - public void setAutoCommit(boolean arg0) throws SQLException + public void setAutoCommit(boolean autoCommit) throws SQLException { - throw new UnsupportedOperationException("method not supported"); - + if (!autoCommit) + throw new SQLException("the Cassandra Implementation is always in auto-commit mode"); } @@ -569,29 +611,35 @@ class CassandraConnection implements Connection */ public void setCatalog(String arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is there are no catalog name to set in this implementation... + // so we are "silently ignoring" the request } /** - * @param arg0 + * @param props * @throws SQLClientInfoException */ - public void setClientInfo(Properties arg0) throws SQLClientInfoException + public void setClientInfo(Properties props) throws SQLClientInfoException { - throw new UnsupportedOperationException("method not supported"); + // this needs to be revisited when and if we actually start to use the clientInfo properties + if (props != null) + clientInfo = props; } /** - * @param arg0 - * @param arg1 + * @param key + * @param value * @throws SQLClientInfoException */ - public void setClientInfo(String arg0, String arg1) throws SQLClientInfoException + public void setClientInfo(String key, String value) throws SQLClientInfoException { - throw new UnsupportedOperationException("method not supported"); - + // this needs to be revisited when and if we actually start to use the clientInfo properties + clientInfo.setProperty(key, value); } @@ -601,9 +649,13 @@ class CassandraConnection implements Connection */ public void setHoldability(int arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is there are no holdability to set in this implementation... + // so we are "silently ignoring" the request } - + /** * @param arg0 @@ -611,9 +663,13 @@ class CassandraConnection implements Connection */ public void setReadOnly(boolean arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + // the rationale is all connections are read/write in the Cassandra implementation... + // so we are "silently ignoring" the request } - + /** * @return @@ -623,7 +679,7 @@ class CassandraConnection implements Connection { throw new UnsupportedOperationException("method not supported"); } - + /** * @param arg0 @@ -632,19 +688,23 @@ class CassandraConnection implements Connection */ public Savepoint setSavepoint(String arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } - + /** - * @param arg0 + * @param level * @throws SQLException */ - public void setTransactionIsolation(int arg0) throws SQLException + public void setTransactionIsolation(int level) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + if (isClosed()) + throw new SQLException("this method was called on a closed Connection"); + + if (level != Connection.TRANSACTION_NONE) + throw new SQLException("the Cassandra Inplementation does not support transactions"); } - + /** * @param arg0 @@ -652,7 +712,7 @@ class CassandraConnection implements Connection */ public void setTypeMap(Map> arg0) throws SQLException { - throw new UnsupportedOperationException("method not supported"); + throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method"); } } diff --git a/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraResultSet.java b/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraResultSet.java index 809db703c7..fac74fd0c1 100644 --- a/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraResultSet.java +++ b/drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraResultSet.java @@ -30,6 +30,7 @@ public interface CassandraResultSet extends ResultSet * @return the current row key */ public byte[] getKey(); + public TypedColumn getTypedKey(); /** @return a BigInteger value for the given column offset*/ public BigInteger getBigInteger(int i); diff --git a/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java b/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java index 7d8663b1d7..a9b15b76f1 100644 --- a/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java +++ b/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java @@ -128,4 +128,14 @@ class ColumnDecoder getNameType(keyspace, columnFamily, column.name), getValueType(keyspace, columnFamily, column.name)); } + + /** constructs a typed column to hold the key */ + public TypedColumn makeKeyColumn(String keyspace, String columnFamily, byte[] key) + { + CFMetaData md = metadata.get(String.format("%s.%s", keyspace, columnFamily)); + Column column = new Column(md.getKeyName()).setValue(key).setTimestamp(-1); + return new TypedColumn(column, + getNameType(keyspace, columnFamily, md.getKeyName()), + getValueType(keyspace, columnFamily, md.getKeyName())); + } } diff --git a/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java b/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java index f3f748fd7b..b48f33febf 100644 --- a/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java +++ b/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java @@ -27,7 +27,10 @@ import java.math.BigInteger; import java.nio.ByteBuffer; import java.sql.*; import java.util.Arrays; +import java.util.Iterator; +import java.util.List; +import org.apache.cassandra.cql.jdbc.CassandraResultSet; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -256,27 +259,31 @@ public class JdbcDriverTest extends EmbeddedServiceBase public void testWithStatement() throws SQLException { Statement stmt = con.createStatement(); - + List keys = Arrays.asList(jsmith); String selectQ = "SELECT 1, 2 FROM JdbcInteger WHERE KEY='" + jsmith + "'"; - checkResultSet(stmt.executeQuery(selectQ), "Int", 1, "1", "2"); + checkResultSet(stmt.executeQuery(selectQ), "Int", 1, keys, "1", "2"); selectQ = "SELECT 3, 4 FROM JdbcInteger WHERE KEY='" + jsmith + "'"; - checkResultSet(stmt.executeQuery(selectQ), "Int", 1, "3", "4"); + checkResultSet(stmt.executeQuery(selectQ), "Int", 1, keys, "3", "4"); selectQ = "SELECT 1, 2, 3, 4 FROM JdbcInteger WHERE KEY='" + jsmith + "'"; - checkResultSet(stmt.executeQuery(selectQ), "Int", 1, "1", "2", "3", "4"); + checkResultSet(stmt.executeQuery(selectQ), "Int", 1, keys, "1", "2", "3", "4"); selectQ = "SELECT 1, 2 FROM JdbcLong WHERE KEY='" + jsmith + "'"; - checkResultSet(stmt.executeQuery(selectQ), "Long", 1, "1", "2"); + checkResultSet(stmt.executeQuery(selectQ), "Long", 1, keys, "1", "2"); selectQ = "SELECT 'first', last FROM JdbcAscii WHERE KEY='" + jsmith + "'"; - checkResultSet(stmt.executeQuery(selectQ), "String", 1, "first", "last"); + checkResultSet(stmt.executeQuery(selectQ), "String", 1, keys, "first", "last"); selectQ = String.format("SELECT '%s', '%s' FROM JdbcBytes WHERE KEY='%s'", first, last, jsmith); - checkResultSet(stmt.executeQuery(selectQ), "Bytes", 1, first, last); + checkResultSet(stmt.executeQuery(selectQ), "Bytes", 1, keys, first, last); selectQ = "SELECT 'first', last FROM JdbcUtf8 WHERE KEY='" + jsmith + "'"; - checkResultSet(stmt.executeQuery(selectQ), "String", 1, "first", "last"); + checkResultSet(stmt.executeQuery(selectQ), "String", 1, keys, "first", "last"); + + String badKey = FBUtilities.bytesToHex(String.format("jsmith-%s", System.currentTimeMillis()).getBytes()); + selectQ = "SELECT 1, 2 FROM JdbcInteger WHERE KEY IN ('" + badKey + "', '" + jsmith + "')"; + checkResultSet(stmt.executeQuery(selectQ), "Int", 1, keys, "1", "2"); } @Test @@ -293,29 +300,35 @@ public class JdbcDriverTest extends EmbeddedServiceBase @Test public void testWithPreparedStatement() throws SQLException { + List keys = Arrays.asList(jsmith); + String selectQ = String.format("SELECT '%s', '%s' FROM Standard1 WHERE KEY='%s'", first, last, jsmith); - checkResultSet(executePreparedStatementWithResults(con, selectQ), "Bytes", 1, first, last); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Bytes", 1, keys, first, last); selectQ = "SELECT 1, 2 FROM JdbcInteger WHERE KEY='" + jsmith + "'"; - checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, "1", "2"); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, keys, "1", "2"); selectQ = "SELECT 3, 4 FROM JdbcInteger WHERE KEY='" + jsmith + "'"; - checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, "3", "4"); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, keys, "3", "4"); selectQ = "SELECT 1, 2, 3, 4 FROM JdbcInteger WHERE KEY='" + jsmith + "'"; - checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, "1", "2", "3", "4"); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, keys, "1", "2", "3", "4"); selectQ = "SELECT 1, 2 FROM JdbcLong WHERE KEY='" + jsmith + "'"; - checkResultSet(executePreparedStatementWithResults(con, selectQ), "Long", 1, "1", "2"); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Long", 1, keys, "1", "2"); selectQ = "SELECT 'first', last FROM JdbcAscii WHERE KEY='" + jsmith + "'"; - checkResultSet(executePreparedStatementWithResults(con, selectQ), "String", 1, "first", "last"); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "String", 1, keys, "first", "last"); selectQ = String.format("SELECT '%s', '%s' FROM JdbcBytes WHERE KEY='%s'", first, last, jsmith); - checkResultSet(executePreparedStatementWithResults(con, selectQ), "Bytes", 1, first, last); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Bytes", 1, keys, first, last); selectQ = "SELECT 'first', last FROM JdbcUtf8 WHERE KEY='" + jsmith + "'"; - checkResultSet(executePreparedStatementWithResults(con, selectQ), "String", 1, "first", "last"); + checkResultSet(executePreparedStatementWithResults(con, selectQ), "String", 1, keys, "first", "last"); + + String badKey = FBUtilities.bytesToHex(String.format("jsmith-%s", System.currentTimeMillis()).getBytes()); + selectQ = "SELECT 1, 2 FROM JdbcInteger WHERE KEY IN ('" + badKey + "', '" + jsmith + "')"; + checkResultSet(executePreparedStatementWithResults(con, selectQ), "Int", 1, keys, "1", "2"); } /* Method to test with Delete statement. */ @@ -411,12 +424,24 @@ public class JdbcDriverTest extends EmbeddedServiceBase // todo: check expected values as well. /** iterates over a result set checking columns */ private static void checkResultSet(ResultSet rs, String accessor, int expectedRows, String... cols) throws SQLException + { + checkResultSet(rs, accessor, expectedRows, null, cols); + } + + private static void checkResultSet(ResultSet rs, String accessor, int expectedRows, List keys, String... cols) throws SQLException { int actualRows = 0; assert rs != null; + Iterator keyIter = (keys == null) ? null : keys.iterator(); + CassandraResultSet cassandraRs = (CassandraResultSet)rs; while (rs.next()) { actualRows++; + if (keyIter != null) + { + assert cassandraRs.getTypedKey().getValueString().equals(keyIter.next()); + } + for (int c = 0; c < cols.length; c++) { // getObject should always work. diff --git a/drivers/py/cql/marshal.py b/drivers/py/cql/marshal.py index 7b69421e49..380c3138fa 100644 --- a/drivers/py/cql/marshal.py +++ b/drivers/py/cql/marshal.py @@ -39,6 +39,7 @@ LONG_TYPE = "org.apache.cassandra.db.marshal.LongType" UUID_TYPE = "org.apache.cassandra.db.marshal.UUIDType" LEXICAL_UUID_TYPE = "org.apache.cassandra.db.marshal.LexicalType" TIME_UUID_TYPE = "org.apache.cassandra.db.marshal.TimeUUIDType" +COUNTER_COLUMN_TYPE = "org.apache.cassandra.db.marshal.CounterColumnType" def prepare(query, params): # For every match of the form ":param_name", call marshal @@ -76,14 +77,15 @@ else: def unmarshal_uuid(bytestr): return UUID(bytes=bytestr) -unmarshallers = {BYTES_TYPE: unmarshal_noop, - ASCII_TYPE: unmarshal_noop, - UTF8_TYPE: unmarshal_utf8, - INTEGER_TYPE: unmarshal_int, - LONG_TYPE: unmarshal_long, - UUID_TYPE: unmarshal_uuid, - LEXICAL_UUID_TYPE: unmarshal_uuid, - TIME_UUID_TYPE: unmarshal_uuid} +unmarshallers = {BYTES_TYPE: unmarshal_noop, + ASCII_TYPE: unmarshal_noop, + UTF8_TYPE: unmarshal_utf8, + INTEGER_TYPE: unmarshal_int, + LONG_TYPE: unmarshal_long, + UUID_TYPE: unmarshal_uuid, + LEXICAL_UUID_TYPE: unmarshal_uuid, + TIME_UUID_TYPE: unmarshal_uuid, + COUNTER_COLUMN_TYPE: unmarshal_long} def decode_bigint(term): val = int(term.encode('hex'), 16) diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 35b9ff9b05..c3fc135c7f 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -257,7 +257,7 @@ public final class CFMetaData public static CFMetaData newIndexMetadata(CFMetaData parent, ColumnDefinition info, AbstractType columnComparator) { - return new CFMetaData(parent.ksName, parent.indexName(info), ColumnFamilyType.Standard, columnComparator, null) + return new CFMetaData(parent.ksName, parent.indexColumnFamilyName(info), ColumnFamilyType.Standard, columnComparator, null) .keyCacheSize(0.0) .readRepairChance(0.0) .gcGraceSeconds(parent.gcGraceSeconds) @@ -305,10 +305,18 @@ public final class CFMetaData cfIdMap.remove(new Pair(cfm.ksName, cfm.cfName)); } - /** convention for nameing secondary indexes. */ - public String indexName(ColumnDefinition info) + /** + * generate a column family name for an index corresponding to the given column. + * This is NOT the same as the index's name! This is only used in sstable filenames and is not exposed to users. + * + * @param info A definition of the column with index + * + * @return name of the index ColumnFamily + */ + public String indexColumnFamilyName(ColumnDefinition info) { - return cfName + "." + (info.getIndexName() == null ? comparator.getString(info.name) + "_idx" : info.getIndexName()); + // TODO simplify this when info.index_name is guaranteed to be set + return cfName + "." + (info.getIndexName() == null ? ByteBufferUtil.bytesToHex(info.name) : info.getIndexName()); } public org.apache.cassandra.db.migration.avro.CfDef deflate() @@ -369,6 +377,8 @@ public final class CFMetaData for (ColumnDef aColumn_metadata : cf.column_metadata) { ColumnDefinition cd = ColumnDefinition.inflate(aColumn_metadata); + if (cd.getIndexName() == null) + cd.setIndexName(getDefaultIndexName(comparator, cd.name)); column_metadata.put(cd.name, cd); } @@ -930,6 +940,37 @@ public final class CFMetaData return column_metadata.get(name); } + /** + * Convert a null index_name to appropriate default name according to column status + * @param cf_def Thrift ColumnFamily Definition + */ + public static void addDefaultIndexNames(org.apache.cassandra.thrift.CfDef cf_def) throws InvalidRequestException + { + if (cf_def.column_metadata == null) + return; + + AbstractType comparator; + try + { + comparator = TypeParser.parse(cf_def.comparator_type); + } + catch (ConfigurationException e) + { + throw new InvalidRequestException(e.getMessage()); + } + + for (org.apache.cassandra.thrift.ColumnDef column : cf_def.column_metadata) + { + if (column.index_type != null && column.index_name == null) + column.index_name = getDefaultIndexName(comparator, column.name); + } + } + + public static String getDefaultIndexName(AbstractType comparator, ByteBuffer columnName) + { + return comparator.getString(columnName).replaceAll("\\W", "") + "_idx"; + } + @Override public String toString() { diff --git a/src/java/org/apache/cassandra/config/KSMetaData.java b/src/java/org/apache/cassandra/config/KSMetaData.java index 2491809cf7..d3a29943f6 100644 --- a/src/java/org/apache/cassandra/config/KSMetaData.java +++ b/src/java/org/apache/cassandra/config/KSMetaData.java @@ -68,7 +68,10 @@ public final class KSMetaData public static Map forwardsCompatibleOptions(KsDef ks_def) { - Map options = new HashMap(ks_def.strategy_options); + Map options; + options = ks_def.strategy_options == null + ? new HashMap() + : new HashMap(ks_def.strategy_options); maybeAddReplicationFactor(options, ks_def.strategy_class, ks_def.isSetReplication_factor() ? ks_def.replication_factor : null); return options; } diff --git a/src/java/org/apache/cassandra/cql/AbstractModification.java b/src/java/org/apache/cassandra/cql/AbstractModification.java index e1c72d2ea3..09afd73e39 100644 --- a/src/java/org/apache/cassandra/cql/AbstractModification.java +++ b/src/java/org/apache/cassandra/cql/AbstractModification.java @@ -20,7 +20,7 @@ */ package org.apache.cassandra.cql; -import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.db.IMutation; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.thrift.InvalidRequestException; @@ -95,8 +95,8 @@ public abstract class AbstractModification * * @throws InvalidRequestException on the wrong request */ - public abstract List prepareRowMutations(String keyspace, ClientState clientState) - throws InvalidRequestException; + public abstract List prepareRowMutations(String keyspace, ClientState clientState) + throws org.apache.cassandra.thrift.InvalidRequestException; /** * Convert statement into a list of mutations to apply on the server @@ -109,37 +109,6 @@ public abstract class AbstractModification * * @throws InvalidRequestException on the wrong request */ - public abstract List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) - throws InvalidRequestException; - - /** - * Compute a row mutation for a single key - * - * @param key The key for mutation - * @param keyspace The keyspace - * @param timestamp The global timestamp for mutation - * - * @return row mutation - * - * @throws InvalidRequestException on the wrong request - */ - public abstract RowMutation mutationForKey(ByteBuffer key, String keyspace, Long timestamp) - throws InvalidRequestException; - - /** - * Compute a row mutation for a single key and add it to the given RowMutation object - * - * @param mutation The row mutation to add computed mutation into - * @param keyspace The keyspace - * @param timestamp The global timestamp for mutation - * - * @throws InvalidRequestException on the wrong request - */ - public abstract void mutationForKey(RowMutation mutation, String keyspace, Long timestamp) - throws InvalidRequestException; - - /** - * @return a list of the keys associated with the statement - */ - public abstract List getKeys(); + public abstract List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) + throws org.apache.cassandra.thrift.InvalidRequestException; } diff --git a/src/java/org/apache/cassandra/cql/AlterTableStatement.java b/src/java/org/apache/cassandra/cql/AlterTableStatement.java new file mode 100644 index 0000000000..89f4e77384 --- /dev/null +++ b/src/java/org/apache/cassandra/cql/AlterTableStatement.java @@ -0,0 +1,122 @@ +/* + * + * 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.cql; + +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.ColumnDefinition; +import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.marshal.TypeParser; +import org.apache.cassandra.db.migration.avro.CfDef; +import org.apache.cassandra.db.migration.avro.ColumnDef; +import org.apache.cassandra.thrift.InvalidRequestException; + +import java.nio.ByteBuffer; + +public class AlterTableStatement +{ + public static enum OperationType + { + ADD, ALTER, DROP + } + + public final OperationType oType; + public final String columnFamily, columnName, validator; + + public AlterTableStatement(String columnFamily, OperationType type, String columnName) + { + this(columnFamily, type, columnName, null); + } + + public AlterTableStatement(String columnFamily, OperationType type, String columnName, String validator) + { + this.columnFamily = columnFamily; + this.oType = type; + this.columnName = columnName; + this.validator = CreateColumnFamilyStatement.comparators.get(validator); // used only for ADD/ALTER commands + } + + public CfDef getCfDef(String keyspace) throws ConfigurationException, InvalidRequestException + { + CFMetaData meta = DatabaseDescriptor.getCFMetaData(keyspace, columnFamily); + + CfDef cfDef = CFMetaData.convertToAvro(meta); + + ByteBuffer columnName = meta.comparator.fromString(this.columnName); + + switch (oType) + { + case ADD: + cfDef.column_metadata.add(new ColumnDefinition(columnName, + TypeParser.parse(validator), + null, + null).deflate()); + break; + + case ALTER: + ColumnDefinition column = meta.getColumnDefinition(columnName); + + if (column == null) + throw new InvalidRequestException(String.format("Column '%s' was not found in CF '%s'", + this.columnName, + columnFamily)); + + column.setValidator(TypeParser.parse(validator)); + + cfDef.column_metadata.add(column.deflate()); + break; + + case DROP: + ColumnDef toDelete = null; + + for (ColumnDef columnDef : cfDef.column_metadata) + { + if (columnDef.name.equals(columnName)) + { + toDelete = columnDef; + } + } + + if (toDelete == null) + throw new InvalidRequestException(String.format("Column '%s' was not found in CF '%s'", + this.columnName, + columnFamily)); + + // it is impossible to use ColumnDefinition.deflate() in remove() method + // it will throw java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.avro.util.Utf8 + // some where deep inside of Avro + cfDef.column_metadata.remove(toDelete); + break; + } + + return cfDef; + } + + public String toString() + { + return String.format("AlterTableStatement(cf=%s, type=%s, column=%s, validator=%s)", + columnFamily, + oType, + columnName, + validator); + } + +} diff --git a/src/java/org/apache/cassandra/cql/BatchStatement.java b/src/java/org/apache/cassandra/cql/BatchStatement.java index 2f84c83985..e1749e1f68 100644 --- a/src/java/org/apache/cassandra/cql/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql/BatchStatement.java @@ -20,23 +20,14 @@ */ package org.apache.cassandra.cql; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.Collection; import java.util.LinkedList; import java.util.List; -import org.apache.cassandra.auth.Permission; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.db.ColumnFamily; -import org.apache.cassandra.db.RowMutation; -import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.IMutation; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.thrift.InvalidRequestException; -import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily; - /** * A BATCH statement parsed from a CQL query. * @@ -85,41 +76,12 @@ public class BatchStatement return timeToLive; } - public List getMutations(String keyspace, ClientState clientState) throws InvalidRequestException + public List getMutations(String keyspace, ClientState clientState) throws InvalidRequestException { - // To avoid unnecessary authorizations. - List seenColumnFamilies = new ArrayList(); + List batch = new LinkedList(); - List batch = new LinkedList(); - - for (AbstractModification statement : statements) - { - final String columnFamily = statement.getColumnFamily(); - - authorizeColumnFamily(keyspace, columnFamily, clientState, seenColumnFamilies); - - AbstractType keyValidator = getKeyType(keyspace, columnFamily); - - for (Term rawKey : statement.getKeys()) // for each key of the statement - { - ByteBuffer key = rawKey.getByteBuffer(keyValidator); - - boolean found = false; - - for (RowMutation mutation : batch) - { - if (mutation.key().equals(key) && hasColumnFamily(mutation.getColumnFamilies(), columnFamily)) - { - statement.mutationForKey(mutation, keyspace, timestamp); - - found = true; - break; - } - } - - if (!found) // if mutation was not found we should add a new one - batch.add(statement.mutationForKey(key, keyspace, timestamp)); - } + for (AbstractModification statement : statements) { + batch.addAll(statement.prepareRowMutations(keyspace, clientState, timestamp)); } return batch; @@ -130,34 +92,6 @@ public class BatchStatement return timestamp != null; } - private boolean hasColumnFamily(Collection columnFamilies, String columnFamily) - { - for (ColumnFamily cf : columnFamilies) - { - if (cf.metadata().cfName.equals(columnFamily)) - return true; - } - - return false; - } - - private void authorizeColumnFamily(String keyspace, String columnFamily, ClientState state, List seenCFs) - throws InvalidRequestException - { - validateColumnFamily(keyspace, columnFamily, false); - - if (!seenCFs.contains(columnFamily)) - { - state.hasColumnFamilyAccess(columnFamily, Permission.WRITE); - seenCFs.add(columnFamily); - } - } - - public AbstractType getKeyType(String keyspace, String columnFamily) - { - return DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator(); - } - public String toString() { return String.format("BatchStatement(statements=%s, consistency=%s)", statements, consistency); diff --git a/src/java/org/apache/cassandra/cql/Cql.g b/src/java/org/apache/cassandra/cql/Cql.g index b4e73609dd..758582ac90 100644 --- a/src/java/org/apache/cassandra/cql/Cql.g +++ b/src/java/org/apache/cassandra/cql/Cql.g @@ -35,6 +35,8 @@ options { import java.util.ArrayList; import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.thrift.InvalidRequestException; + + import static org.apache.cassandra.cql.AlterTableStatement.OperationType; } @members { @@ -111,8 +113,10 @@ query returns [CQLStatement stmnt] | createKeyspaceStatement { $stmnt = new CQLStatement(StatementType.CREATE_KEYSPACE, $createKeyspaceStatement.expr); } | createColumnFamilyStatement { $stmnt = new CQLStatement(StatementType.CREATE_COLUMNFAMILY, $createColumnFamilyStatement.expr); } | createIndexStatement { $stmnt = new CQLStatement(StatementType.CREATE_INDEX, $createIndexStatement.expr); } + | dropIndexStatement { $stmnt = new CQLStatement(StatementType.DROP_INDEX, $dropIndexStatement.expr); } | dropKeyspaceStatement { $stmnt = new CQLStatement(StatementType.DROP_KEYSPACE, $dropKeyspaceStatement.ksp); } | dropColumnFamilyStatement { $stmnt = new CQLStatement(StatementType.DROP_COLUMNFAMILY, $dropColumnFamilyStatement.cfam); } + | alterTableStatement { $stmnt = new CQLStatement(StatementType.ALTER_TABLE, $alterTableStatement.expr); } ; // USE ; @@ -288,12 +292,12 @@ batchStatementObjective returns [AbstractModification statement] updateStatement returns [UpdateStatement expr] : { Attributes attrs = new Attributes(); - Map columns = new HashMap(); + Map columns = new HashMap(); List keyList = null; } K_UPDATE columnFamily=( IDENT | STRING_LITERAL | INTEGER ) ( usingClause[attrs] )? - K_SET termPair[columns] (',' termPair[columns])* + K_SET termPairWithOperation[columns] (',' termPairWithOperation[columns])* K_WHERE ( K_KEY '=' key=term { keyList = Collections.singletonList(key); } | K_KEY K_IN '(' keys=termList { keyList = $keys.items; } ')' ) @@ -380,19 +384,49 @@ createIndexStatement returns [CreateIndexStatement expr] : K_CREATE K_INDEX (idxName=IDENT)? K_ON cf=( IDENT | STRING_LITERAL | INTEGER ) '(' columnName=term ')' endStmnt { $expr = new CreateIndexStatement($idxName.text, $cf.text, columnName); } ; +/** + * DROP INDEX ON . + * DROP INDEX + */ +dropIndexStatement returns [DropIndexStatement expr] + : + K_DROP K_INDEX index=( IDENT | STRING_LITERAL | INTEGER ) endStmnt + { $expr = new DropIndexStatement($index.text); } + ; /** DROP KEYSPACE ; */ dropKeyspaceStatement returns [String ksp] : K_DROP K_KEYSPACE name=( IDENT | STRING_LITERAL | INTEGER ) endStmnt { $ksp = $name.text; } ; + +alterTableStatement returns [AlterTableStatement expr] + : + { + OperationType type = null; + String columnFamily = null, columnName = null, validator = null; + } + K_ALTER K_TABLE name=( IDENT | STRING_LITERAL | INTEGER ) { columnFamily = $name.text; } + ( K_ALTER { type = OperationType.ALTER; } + (col=( IDENT | STRING_LITERAL | INTEGER ) { columnName = $col.text; }) + K_TYPE alterValidator=comparatorType { validator = $alterValidator.text; } + | K_ADD { type = OperationType.ADD; } + (col=( IDENT | STRING_LITERAL | INTEGER ) { columnName = $col.text; }) + addValidator=comparatorType { validator = $addValidator.text; } + | K_DROP { type = OperationType.DROP; } + (col=( IDENT | STRING_LITERAL | INTEGER ) { columnName = $col.text; })) + endStmnt + { + $expr = new AlterTableStatement(columnFamily, type, columnName, validator); + } + ; /** DROP COLUMNFAMILY ; */ dropColumnFamilyStatement returns [String cfam] : K_DROP K_COLUMNFAMILY name=( IDENT | STRING_LITERAL | INTEGER ) endStmnt { $cfam = $name.text; } ; comparatorType - : 'bytea' | 'ascii' | 'text' | 'varchar' | 'int' | 'varint' | 'bigint' | 'uuid' + : 'bytea' | 'ascii' | 'text' | 'varchar' | 'int' | 'varint' | 'bigint' | 'uuid' | 'counter' ; term returns [Term item] @@ -409,6 +443,12 @@ termPair[Map columns] : key=term '=' value=term { columns.put(key, value); } ; +termPairWithOperation[Map columns] + : key=term '=' (value=term { columns.put(key, new Operation(value)); } + | c=term ( '+' v=term { columns.put(key, new Operation(c, org.apache.cassandra.cql.Operation.OperationType.PLUS, v)); } + | '-' v=term { columns.put(key, new Operation(c, org.apache.cassandra.cql.Operation.OperationType.MINUS, v)); } )) + ; + // Note: ranges are inclusive so >= and >, and < and <= all have the same semantics. relation returns [Relation rel] : { Term entity = new Term("KEY", STRING_LITERAL); } @@ -468,6 +508,10 @@ K_INTO: I N T O; K_VALUES: V A L U E S; K_TIMESTAMP: T I M E S T A M P; K_TTL: T T L; +K_ALTER: A L T E R; +K_TABLE: T A B L E; +K_ADD: A D D; +K_TYPE: T Y P E; // Case-insensitive alpha characters fragment A: ('a'|'A'); diff --git a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java index 73ce899a76..86b2c65e99 100644 --- a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java +++ b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java @@ -58,7 +58,7 @@ public class CreateColumnFamilyStatement private static final String KW_ROW_CACHE_PROVIDER = "row_cache_provider"; // Maps CQL short names to the respective Cassandra comparator/validator class names - private static final Map comparators = new HashMap(); + public static final Map comparators = new HashMap(); private static final Set keywords = new HashSet(); static @@ -71,6 +71,7 @@ public class CreateColumnFamilyStatement comparators.put("int", "LongType"); comparators.put("bigint", "LongType"); comparators.put("uuid", "UUIDType"); + comparators.put("counter", "CounterColumnType"); keywords.add(KW_COMPARATOR); keywords.add(KW_COMMENT); diff --git a/src/java/org/apache/cassandra/cql/DeleteStatement.java b/src/java/org/apache/cassandra/cql/DeleteStatement.java index 120cb4475c..ad08059ae6 100644 --- a/src/java/org/apache/cassandra/cql/DeleteStatement.java +++ b/src/java/org/apache/cassandra/cql/DeleteStatement.java @@ -27,6 +27,7 @@ import java.util.List; import org.apache.cassandra.auth.Permission; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.IMutation; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; @@ -66,18 +67,18 @@ public class DeleteStatement extends AbstractModification } /** {@inheritDoc} */ - public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException + public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException { return prepareRowMutations(keyspace, clientState, null); } /** {@inheritDoc} */ - public List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException + public List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException { clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE); AbstractType keyType = DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator(); - List rowMutations = new ArrayList(); + List rowMutations = new ArrayList(); for (Term key : keys) { @@ -100,7 +101,8 @@ public class DeleteStatement extends AbstractModification /** {@inheritDoc} */ public void mutationForKey(RowMutation mutation, String keyspace, Long timestamp) throws InvalidRequestException { - CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false); + CFMetaData metadata = validateColumnFamily(keyspace, columnFamily); + AbstractType comparator = metadata.getComparatorFor(null); if (columns.size() < 1) // No columns, delete the row diff --git a/src/java/org/apache/cassandra/cql/Operation.java b/src/java/org/apache/cassandra/cql/Operation.java new file mode 100644 index 0000000000..661495c44b --- /dev/null +++ b/src/java/org/apache/cassandra/cql/Operation.java @@ -0,0 +1,58 @@ +/* + * + * 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.cql; + +public class Operation +{ + public static enum OperationType + { PLUS, MINUS } + + public final OperationType type; + public final Term a, b; + + // unary operation + public Operation(Term a) + { + this.a = a; + type = null; + b = null; + } + + // binary operation + public Operation(Term a, OperationType type, Term b) + { + this.a = a; + this.type = type; + this.b = b; + } + + public boolean isUnary() + { + return type == null && b == null; + } + + public String toString() + { + return (isUnary()) + ? String.format("UnaryOperation(%s)", a) + : String.format("BinaryOperation(%s, %s, %s)", a, type, b); + } +} diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java index a33fb97736..d1c886749e 100644 --- a/src/java/org/apache/cassandra/cql/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java @@ -32,6 +32,8 @@ import java.util.concurrent.TimeoutException; import com.google.common.base.Predicates; import com.google.common.collect.Maps; +import org.apache.cassandra.db.CounterColumn; +import org.apache.cassandra.db.context.CounterContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -45,10 +47,8 @@ import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.db.migration.*; -import org.apache.cassandra.db.migration.avro.CfDef; import org.apache.cassandra.dht.*; import org.apache.cassandra.dht.Token; -import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.StorageProxy; import org.apache.cassandra.service.StorageService; @@ -67,7 +67,7 @@ public class QueryProcessor throws InvalidRequestException, TimedOutException, UnavailableException { QueryPath queryPath = new QueryPath(select.getColumnFamily()); - CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false); + CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily()); List commands = new ArrayList(); // ...of a list of column names @@ -161,7 +161,7 @@ public class QueryProcessor } AbstractBounds bounds = new Bounds(startToken, finishToken); - CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false); + CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily()); // XXX: Our use of Thrift structs internally makes me Sad. :( SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, metadata); validateSlicePredicate(metadata, thriftSlicePredicate); @@ -214,7 +214,7 @@ public class QueryProcessor private static List getIndexedSlices(String keyspace, SelectStatement select) throws TimedOutException, UnavailableException, InvalidRequestException { - CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false); + CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily()); // XXX: Our use of Thrift structs internally (still) makes me Sad. :~( SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, metadata); validateSlicePredicate(metadata, thriftSlicePredicate); @@ -261,7 +261,7 @@ public class QueryProcessor throws InvalidRequestException, UnavailableException, TimedOutException { String keyspace = clientState.getKeyspace(); - List rowMutations = new ArrayList(); + List rowMutations = new ArrayList(); List cfamsSeen = new ArrayList(); for (UpdateStatement update : updateStatements) @@ -491,7 +491,7 @@ public class QueryProcessor case SELECT: SelectStatement select = (SelectStatement)statement.statement; clientState.hasColumnFamilyAccess(select.getColumnFamily(), Permission.READ); - metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false); + metadata = validateColumnFamily(keyspace, select.getColumnFamily()); validateSelect(keyspace, select); List rows; @@ -640,7 +640,7 @@ public class QueryProcessor { KsDef ksd = new KsDef(create.getName(), create.getStrategyClass(), - Collections.emptyList()) + Collections.emptyList()) .setStrategy_options(create.getStrategyOptions()); ThriftValidation.validateKsDef(ksd); applyMigrationOnStage(new AddKeyspace(KSMetaData.fromThrift(ksd))); @@ -694,37 +694,33 @@ public class QueryProcessor createIdx.getColumnFamily())); if (oldCfm == null) throw new InvalidRequestException("No such column family: " + createIdx.getColumnFamily()); - + + boolean columnExists = false; ByteBuffer columnName = createIdx.getColumnName().getByteBuffer(); - ColumnDefinition columnDef = oldCfm.getColumn_metadata().get(columnName); - - // Meta-data for this column already exists - if (columnDef != null) + // mutating oldCfm directly would be bad, but mutating a Thrift copy is fine. This also + // sets us up to use validateCfDef to check for index name collisions. + CfDef cf_def = CFMetaData.convertToThrift(oldCfm); + for (ColumnDef cd : cf_def.column_metadata) { - // This column is already indexed, stop, drop, and roll. - if (columnDef.getIndexType() != null) - throw new InvalidRequestException("Index exists"); - // Add index attrs to the existing definition - columnDef.setIndexName(createIdx.getIndexName()); - columnDef.setIndexType(org.apache.cassandra.thrift.IndexType.KEYS); + if (cd.name.equals(columnName)) + { + if (cd.index_type != null) + throw new InvalidRequestException("Index already exists"); + logger.debug("Updating column {} definition for index {}", oldCfm.comparator.getString(columnName), createIdx.getIndexName()); + cd.setIndex_type(IndexType.KEYS); + cd.setIndex_name(createIdx.getIndexName()); + columnExists = true; + break; + } } - // No meta-data, create a new column definition from scratch. - else - { - columnDef = new ColumnDefinition(columnName, - DatabaseDescriptor.getValueValidator(keyspace, - createIdx.getColumnFamily(), - columnName), - org.apache.cassandra.thrift.IndexType.KEYS, - createIdx.getIndexName()); - } - - CfDef cfamilyDef = CFMetaData.convertToAvro(oldCfm); - cfamilyDef.column_metadata.add(columnDef.deflate()); - + if (!columnExists) + throw new InvalidRequestException("No column definition found for column " + oldCfm.comparator.getString(columnName)); + + CFMetaData.addDefaultIndexNames(cf_def); + ThriftValidation.validateCfDef(cf_def, oldCfm); try { - applyMigrationOnStage(new UpdateColumnFamily(cfamilyDef)); + applyMigrationOnStage(new UpdateColumnFamily(CFMetaData.convertToAvro(cf_def))); } catch (ConfigurationException e) { @@ -741,7 +737,32 @@ public class QueryProcessor result.type = CqlResultType.VOID; return result; - + + case DROP_INDEX: + DropIndexStatement dropIdx = (DropIndexStatement)statement.statement; + clientState.hasColumnFamilyListAccess(Permission.WRITE); + validateSchemaAgreement(); + + try + { + applyMigrationOnStage(dropIdx.generateMutation(clientState.getKeyspace())); + } + catch (ConfigurationException e) + { + InvalidRequestException ex = new InvalidRequestException(e.toString()); + ex.initCause(e); + throw ex; + } + catch (IOException e) + { + InvalidRequestException ex = new InvalidRequestException(e.toString()); + ex.initCause(e); + throw ex; + } + + result.type = CqlResultType.VOID; + return result; + case DROP_KEYSPACE: String deleteKeyspace = (String)statement.statement; clientState.hasKeyspaceListAccess(Permission.WRITE); @@ -791,7 +812,35 @@ public class QueryProcessor result.type = CqlResultType.VOID; return result; - + + case ALTER_TABLE: + AlterTableStatement alterTable = (AlterTableStatement) statement.statement; + + System.out.println(alterTable); + + validateColumnFamily(keyspace, alterTable.columnFamily); + clientState.hasColumnFamilyAccess(alterTable.columnFamily, Permission.WRITE); + validateSchemaAgreement(); + + try + { + applyMigrationOnStage(new UpdateColumnFamily(alterTable.getCfDef(keyspace))); + } + catch (ConfigurationException e) + { + InvalidRequestException ex = new InvalidRequestException(e.getMessage()); + ex.initCause(e); + throw ex; + } + catch (IOException e) + { + InvalidRequestException ex = new InvalidRequestException(e.getMessage()); + ex.initCause(e); + throw ex; + } + + result.type = CqlResultType.VOID; + return result; } return null; // We should never get here. @@ -813,7 +862,7 @@ public class QueryProcessor { if (c.isMarkedForDelete()) continue; - thriftColumns.add(new Column(c.name()).setValue(c.value()).setTimestamp(c.timestamp())); + thriftColumns.add(thriftify(c)); } } else @@ -840,16 +889,25 @@ public class QueryProcessor { throw new AssertionError(e); } + IColumn c = row.cf.getColumn(name); if (c == null || c.isMarkedForDelete()) thriftColumns.add(new Column().setName(name)); else - thriftColumns.add(new Column(c.name()).setValue(c.value()).setTimestamp(c.timestamp())); + thriftColumns.add(thriftify(c)); } } return thriftColumns; } + private static Column thriftify(IColumn c) + { + ByteBuffer value = (c instanceof CounterColumn) + ? ByteBufferUtil.bytes(CounterContext.instance().total(c.value())) + : c.value(); + return new Column(c.name()).setValue(value).setTimestamp(c.timestamp()); + } + private static String getKeyString(CFMetaData metadata) { String keyString; diff --git a/src/java/org/apache/cassandra/cql/StatementType.java b/src/java/org/apache/cassandra/cql/StatementType.java index 330f89a302..7988654653 100644 --- a/src/java/org/apache/cassandra/cql/StatementType.java +++ b/src/java/org/apache/cassandra/cql/StatementType.java @@ -24,8 +24,8 @@ import java.util.EnumSet; public enum StatementType { - SELECT, INSERT, UPDATE, BATCH, USE, TRUNCATE, DELETE, CREATE_KEYSPACE, CREATE_COLUMNFAMILY, CREATE_INDEX, - DROP_KEYSPACE, DROP_COLUMNFAMILY; + SELECT, INSERT, UPDATE, BATCH, USE, TRUNCATE, DELETE, CREATE_KEYSPACE, CREATE_COLUMNFAMILY, CREATE_INDEX, DROP_INDEX, + DROP_KEYSPACE, DROP_COLUMNFAMILY, ALTER_TABLE; // Statement types that don't require a keyspace to be set. private static final EnumSet topLevel = EnumSet.of(USE, CREATE_KEYSPACE, DROP_KEYSPACE); diff --git a/src/java/org/apache/cassandra/cql/UpdateStatement.java b/src/java/org/apache/cassandra/cql/UpdateStatement.java index f2e6de890b..ddf4706d5b 100644 --- a/src/java/org/apache/cassandra/cql/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql/UpdateStatement.java @@ -26,6 +26,8 @@ import java.util.*; import org.apache.cassandra.auth.Permission; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.CounterMutation; +import org.apache.cassandra.db.IMutation; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; @@ -35,6 +37,7 @@ import org.apache.cassandra.thrift.InvalidRequestException; import static org.apache.cassandra.cql.QueryProcessor.validateColumn; +import static org.apache.cassandra.cql.Operation.OperationType; import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily; /** @@ -43,7 +46,7 @@ import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily; */ public class UpdateStatement extends AbstractModification { - private Map columns; + private Map columns; private List columnNames, columnValues; private List keys; @@ -57,7 +60,7 @@ public class UpdateStatement extends AbstractModification * @param attrs additional attributes for statement (CL, timestamp, timeToLive) */ public UpdateStatement(String columnFamily, - Map columns, + Map columns, List keys, Attributes attrs) { @@ -113,17 +116,28 @@ public class UpdateStatement extends AbstractModification } /** {@inheritDoc} */ - public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException + public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException { return prepareRowMutations(keyspace, clientState, null); } /** {@inheritDoc} */ - public List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException + public List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException { List cfamsSeen = new ArrayList(); - CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false); + boolean hasCommutativeOperation = false; + + for (Map.Entry column : getColumns().entrySet()) + { + if (!column.getValue().isUnary()) + hasCommutativeOperation = true; + + if (hasCommutativeOperation && column.getValue().isUnary()) + throw new InvalidRequestException("Mix of commutative and non-commutative operations is not allowed."); + } + + CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, hasCommutativeOperation); // Avoid unnecessary authorizations. if (!(cfamsSeen.contains(columnFamily))) @@ -132,7 +146,7 @@ public class UpdateStatement extends AbstractModification cfamsSeen.add(columnFamily); } - List rowMutations = new LinkedList(); + List rowMutations = new LinkedList(); for (Term key: keys) { @@ -154,43 +168,61 @@ public class UpdateStatement extends AbstractModification * * @throws InvalidRequestException on the wrong request */ - private RowMutation mutationForKey(String keyspace, ByteBuffer key, CFMetaData metadata, Long timestamp) throws InvalidRequestException - { - RowMutation rm = new RowMutation(keyspace, key); - - mutationForKey(rm, keyspace, metadata, timestamp); - - return rm; - } - - /** {@inheritDoc} */ - public RowMutation mutationForKey(ByteBuffer key, String keyspace, Long timestamp) throws InvalidRequestException - { - return mutationForKey(keyspace, key, validateColumnFamily(keyspace, columnFamily, false), timestamp); - } - - /** {@inheritDoc} */ - public void mutationForKey(RowMutation mutation, String keyspace, Long timestamp) throws InvalidRequestException - { - mutationForKey(mutation, keyspace, validateColumnFamily(keyspace, columnFamily, false), timestamp); - } - - private void mutationForKey(RowMutation mutation, String keyspace, CFMetaData metadata, Long timestamp) throws InvalidRequestException + private IMutation mutationForKey(String keyspace, ByteBuffer key, CFMetaData metadata, Long timestamp) throws InvalidRequestException { AbstractType comparator = getComparator(keyspace); - for (Map.Entry column : getColumns().entrySet()) + // if true we need to wrap RowMutation into CounterMutation + boolean hasCounterColumn = false; + RowMutation rm = new RowMutation(keyspace, key); + + for (Map.Entry column : getColumns().entrySet()) { ByteBuffer colName = column.getKey().getByteBuffer(comparator); - ByteBuffer colValue = column.getValue().getByteBuffer(getValueValidator(keyspace, colName)); + Operation op = column.getValue(); - validateColumn(metadata, colName, colValue); + if (op.isUnary()) + { + if (hasCounterColumn) + throw new InvalidRequestException("Mix of commutative and non-commutative operations is not allowed."); - mutation.add(new QueryPath(columnFamily, null, colName), - colValue, - (timestamp == null) ? getTimestamp() : timestamp, - getTimeToLive()); + ByteBuffer colValue = op.a.getByteBuffer(getValueValidator(keyspace, colName)); + + validateColumn(metadata, colName, colValue); + rm.add(new QueryPath(columnFamily, null, colName), + colValue, + (timestamp == null) ? getTimestamp() : timestamp, + getTimeToLive()); + } + else + { + hasCounterColumn = true; + + if (!column.getKey().getText().equals(op.a.getText())) + throw new InvalidRequestException("Only expressions like X = X + are supported."); + + long value; + + try + { + value = Long.parseLong(op.b.getText()); + + if (op.type == OperationType.MINUS) + { + value *= -1; + } + } + catch (NumberFormatException e) + { + throw new InvalidRequestException(String.format("'%s' is an invalid value, should be a long.", + op.b.getText())); + } + + rm.addCounter(new QueryPath(columnFamily, null, colName), value); + } } + + return (hasCounterColumn) ? new CounterMutation(rm, getConsistencyLevel()) : rm; } public String getColumnFamily() @@ -203,8 +235,8 @@ public class UpdateStatement extends AbstractModification { return keys; } - - public Map getColumns() throws InvalidRequestException + + public Map getColumns() throws InvalidRequestException { // Created from an UPDATE if (columns != null) @@ -218,11 +250,11 @@ public class UpdateStatement extends AbstractModification if (columnNames.size() < 1) throw new InvalidRequestException("no columns specified for INSERT"); - columns = new HashMap(); + columns = new HashMap(); for (int i = 0; i < columnNames.size(); i++) - columns.put(columnNames.get(i), columnValues.get(i)); - + columns.put(columnNames.get(i), new Operation(columnValues.get(i))); + return columns; } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 6eb3169bf3..d56eea7a51 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -511,7 +511,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean if (cfm != null) // secondary indexes aren't stored in DD. { for (ColumnDefinition def : cfm.getColumn_metadata().values()) - scrubDataDirectories(table, cfm.indexName(def)); + scrubDataDirectories(table, cfm.indexColumnFamilyName(def)); } } @@ -1802,8 +1802,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean */ public Future truncate() throws IOException { - // snapshot will also flush, but we want to truncate the most possible, and anything in a flush written - // after truncateAt won't be truncated. + // We have two goals here: + // - truncate should delete everything written before truncate was invoked + // - but not delete anything that isn't part of the snapshot we create. + // We accomplish this by first flushing manually, then snapshotting, and + // recording the timestamp IN BETWEEN those actions. Any sstables created + // with this timestamp or greater time, will not be marked for delete. try { forceBlockingFlush(); @@ -1812,33 +1816,20 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { throw new RuntimeException(e); } - - final long truncatedAt = System.currentTimeMillis(); + // sleep a little to make sure that our truncatedAt comes after any sstable + // that was part of the flushed we forced; otherwise on a tie, it won't get deleted. + try + { + Thread.sleep(100); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + long truncatedAt = System.currentTimeMillis(); snapshot(Table.getTimestampedSnapshotName("before-truncate")); - Runnable runnable = new WrappedRunnable() - { - public void runMayThrow() throws InterruptedException, IOException - { - // putting markCompacted on the commitlogUpdater thread ensures it will run - // after any compactions that were in progress when truncate was called, are finished - for (ColumnFamilyStore cfs : concatWithIndexes()) - { - List truncatedSSTables = new ArrayList(); - for (SSTableReader sstable : cfs.getSSTables()) - { - if (!sstable.newSince(truncatedAt)) - truncatedSSTables.add(sstable); - } - cfs.data.markCompacted(truncatedSSTables); - } - - // Invalidate row cache - invalidateRowCache(); - } - }; - - return postFlushExecutor.submit(runnable); + return CompactionManager.instance.submitTruncate(this, truncatedAt); } // if this errors out, we are in a world of hurt. diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index a76ba8bd46..4d3fc3cff1 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -47,6 +47,7 @@ import org.apache.cassandra.io.*; import org.apache.cassandra.io.sstable.*; import org.apache.cassandra.io.util.BufferedRandomAccessFile; import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.io.util.MemoryInputStream; import org.apache.cassandra.service.AntiEntropyService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.OperationType; @@ -1149,6 +1150,30 @@ public class CompactionManager implements CompactionManagerMBean return executor.submit(runnable); } + public Future submitTruncate(final ColumnFamilyStore main, final long truncatedAt) + { + Runnable runnable = new WrappedRunnable() + { + public void runMayThrow() throws InterruptedException, IOException + { + for (ColumnFamilyStore cfs : main.concatWithIndexes()) + { + List truncatedSSTables = new ArrayList(); + for (SSTableReader sstable : cfs.getSSTables()) + { + if (!sstable.newSince(truncatedAt)) + truncatedSSTables.add(sstable); + } + cfs.markCompacted(truncatedSSTables); + } + + main.invalidateRowCache(); + } + }; + + return executor.submit(runnable); + } + private static int getDefaultGcBefore(ColumnFamilyStore cfs) { return cfs.isIndex() diff --git a/src/java/org/apache/cassandra/db/DataTracker.java b/src/java/org/apache/cassandra/db/DataTracker.java index 3955b4a2c9..4096f575fb 100644 --- a/src/java/org/apache/cassandra/db/DataTracker.java +++ b/src/java/org/apache/cassandra/db/DataTracker.java @@ -22,20 +22,17 @@ package org.apache.cassandra.db; import java.io.File; import java.io.IOError; import java.io.IOException; -import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; -import com.google.common.collect.Iterables; -import org.apache.commons.collections.set.UnmodifiableSet; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Sets; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.cache.AutoSavingCache; import org.apache.cassandra.config.DatabaseDescriptor; - import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.utils.Pair; @@ -451,18 +448,17 @@ public class DataTracker public final Set sstables; public final Set compacting; - public View(Memtable memtable, Set pendingFlush, Set sstables, Set compacting) + View(Memtable memtable, Set pendingFlush, Set sstables, Set compacting) { this.memtable = memtable; - this.memtablesPendingFlush = pendingFlush instanceof UnmodifiableSet ? pendingFlush : Collections.unmodifiableSet(pendingFlush); - this.sstables = sstables instanceof UnmodifiableSet ? sstables : Collections.unmodifiableSet(sstables); - this.compacting = compacting instanceof UnmodifiableSet ? compacting : Collections.unmodifiableSet(compacting); + this.memtablesPendingFlush = pendingFlush; + this.sstables = sstables; + this.compacting = compacting; } public View switchMemtable(Memtable newMemtable) { - Set newPending = new HashSet(memtablesPendingFlush); - newPending.add(memtable); + Set newPending = ImmutableSet.builder().addAll(memtablesPendingFlush).add(memtable).build(); return new View(newMemtable, newPending, sstables, compacting); } @@ -473,32 +469,27 @@ public class DataTracker public View replaceFlushed(Memtable flushedMemtable, SSTableReader newSSTable) { - Set newPendings = new HashSet(memtablesPendingFlush); - Set newSSTables = new HashSet(sstables); - newPendings.remove(flushedMemtable); - newSSTables.add(newSSTable); - return new View(memtable, newPendings, newSSTables, compacting); + Set newPending = ImmutableSet.copyOf(Sets.difference(memtablesPendingFlush, Collections.singleton(flushedMemtable))); + Set newSSTables = ImmutableSet.builder().addAll(sstables).add(newSSTable).build(); + return new View(memtable, newPending, newSSTables, compacting); } public View replace(Collection oldSSTables, Iterable replacements) { - Set sstablesNew = new HashSet(sstables); - Iterables.addAll(sstablesNew, replacements); - sstablesNew.removeAll(oldSSTables); - return new View(memtable, memtablesPendingFlush, sstablesNew, compacting); + Sets.SetView remaining = Sets.difference(sstables, ImmutableSet.copyOf(oldSSTables)); + Set newSSTables = ImmutableSet.builder().addAll(remaining).addAll(replacements).build(); + return new View(memtable, memtablesPendingFlush, newSSTables, compacting); } public View markCompacting(Collection tomark) { - Set compactingNew = new HashSet(compacting); - compactingNew.addAll(tomark); + Set compactingNew = ImmutableSet.builder().addAll(sstables).addAll(tomark).build(); return new View(memtable, memtablesPendingFlush, sstables, compactingNew); } public View unmarkCompacting(Collection tounmark) { - Set compactingNew = new HashSet(compacting); - compactingNew.removeAll(tounmark); + Set compactingNew = ImmutableSet.copyOf(Sets.difference(compacting, ImmutableSet.copyOf(tounmark))); return new View(memtable, memtablesPendingFlush, sstables, compactingNew); } } diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index b72344b95b..1a6e3fb6df 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -731,23 +731,6 @@ public class Table return Iterables.transform(DatabaseDescriptor.getTables(), transformer); } - /** - * Performs a synchronous truncate operation, effectively deleting all data - * from the column family cfname - * @param cfname - * @throws IOException - * @throws ExecutionException - * @throws InterruptedException - */ - public void truncate(String cfname) throws InterruptedException, ExecutionException, IOException - { - logger.debug("Truncating..."); - ColumnFamilyStore cfs = getColumnFamilyStore(cfname); - // truncate, blocking - cfs.truncate().get(); - logger.debug("Truncation done."); - } - @Override public String toString() { diff --git a/src/java/org/apache/cassandra/db/TruncateVerbHandler.java b/src/java/org/apache/cassandra/db/TruncateVerbHandler.java index 83161b0d82..7a7796d67c 100644 --- a/src/java/org/apache/cassandra/db/TruncateVerbHandler.java +++ b/src/java/org/apache/cassandra/db/TruncateVerbHandler.java @@ -22,7 +22,6 @@ import java.io.ByteArrayInputStream; import java.io.DataInputStream; import java.io.IOError; import java.io.IOException; -import java.util.concurrent.ExecutionException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -47,7 +46,8 @@ public class TruncateVerbHandler implements IVerbHandler try { - Table.open(t.keyspace).truncate(t.columnFamily); + ColumnFamilyStore cfs = Table.open(t.keyspace).getColumnFamilyStore(t.columnFamily); + cfs.truncate().get(); } catch (Exception e) { diff --git a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java index 3a1377cf3f..e849e7bcd9 100644 --- a/src/java/org/apache/cassandra/net/IncomingTcpConnection.java +++ b/src/java/org/apache/cassandra/net/IncomingTcpConnection.java @@ -38,6 +38,8 @@ public class IncomingTcpConnection extends Thread { private static Logger logger = LoggerFactory.getLogger(IncomingTcpConnection.class); + private static final int CHUNK_SIZE = 1024 * 1024; + private Socket socket; public IncomingTcpConnection(Socket socket) @@ -97,8 +99,13 @@ public class IncomingTcpConnection extends Thread { int size = input.readInt(); byte[] contentBytes = new byte[size]; - input.readFully(contentBytes); - + // readFully allocates a direct buffer the size of the chunk it is asked to read, + // so we cap that at CHUNK_SIZE. See https://issues.apache.org/jira/browse/CASSANDRA-2654 + int remainder = size % CHUNK_SIZE; + for (int offset = 0; offset < size - remainder; offset += CHUNK_SIZE) + input.readFully(contentBytes, offset, CHUNK_SIZE); + input.readFully(contentBytes, size - remainder, remainder); + if (version > MessagingService.version_) logger.info("Received connection from newer protocol version. Ignorning message."); else diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index ac6074e3fb..4baf84804d 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -799,7 +799,8 @@ public class CassandraServer implements Cassandra.Iface { logger.debug("add_column_family"); state().hasColumnFamilyListAccess(Permission.WRITE); - ThriftValidation.validateCfDef(cf_def); + CFMetaData.addDefaultIndexNames(cf_def); + ThriftValidation.validateCfDef(cf_def, null); validateSchemaAgreement(); try @@ -866,10 +867,11 @@ public class CassandraServer implements Cassandra.Iface try { Collection cfDefs = new ArrayList(ks_def.cf_defs.size()); - for (CfDef cfDef : ks_def.cf_defs) + for (CfDef cf_def : ks_def.cf_defs) { - ThriftValidation.validateCfDef(cfDef); - cfDefs.add(CFMetaData.fromThrift(cfDef)); + CFMetaData.addDefaultIndexNames(cf_def); + ThriftValidation.validateCfDef(cf_def, null); + cfDefs.add(CFMetaData.fromThrift(cf_def)); } ThriftValidation.validateKsDef(ks_def); @@ -953,11 +955,10 @@ public class CassandraServer implements Cassandra.Iface { logger.debug("update_column_family"); state().hasColumnFamilyListAccess(Permission.WRITE); - ThriftValidation.validateCfDef(cf_def); if (cf_def.keyspace == null || cf_def.name == null) throw new InvalidRequestException("Keyspace and CF name must be set."); CFMetaData oldCfm = DatabaseDescriptor.getCFMetaData(CFMetaData.getId(cf_def.keyspace, cf_def.name)); - if (oldCfm == null) + if (oldCfm == null) throw new InvalidRequestException("Could not find column family definition to modify."); validateSchemaAgreement(); diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java index bdd0acf5b0..b0abbe7879 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java +++ b/src/java/org/apache/cassandra/thrift/ThriftValidation.java @@ -29,6 +29,7 @@ import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.AsciiType; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.db.marshal.TypeParser; +import org.apache.cassandra.db.migration.Migration; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.RandomPartitioner; import org.apache.cassandra.dht.Token; @@ -513,7 +514,7 @@ public class ThriftValidation throw new InvalidRequestException("No indexed columns present in index clause with operator EQ"); } - public static void validateCfDef(CfDef cf_def) throws InvalidRequestException + public static void validateCfDef(CfDef cf_def, CFMetaData old) throws InvalidRequestException { try { @@ -533,6 +534,22 @@ public class ThriftValidation } } + if (cf_def.key_alias != null) + { + if (!cf_def.key_alias.hasRemaining()) + throw new InvalidRequestException("key_alias may not be empty"); + try + { + // it's hard to use a key in a select statement if we can't type it. + // for now let's keep it simple and require ascii. + AsciiType.instance.validate(cf_def.key_alias); + } + catch (MarshalException e) + { + throw new InvalidRequestException("Key aliases must be ascii"); + } + } + ColumnFamilyType cfType = ColumnFamilyType.create(cf_def.column_type); if (cfType == null) throw new InvalidRequestException("invalid column type " + cf_def.column_type); @@ -550,16 +567,17 @@ public class ThriftValidation ? TypeParser.parse(cf_def.comparator_type) : TypeParser.parse(cf_def.subcomparator_type); + // initialize a set of names NOT in the CF under consideration Set indexNames = new HashSet(); + for (ColumnFamilyStore cfs : ColumnFamilyStore.all()) + { + if (!cfs.getColumnFamilyName().equals(cf_def.name)) + for (ColumnDefinition cd : cfs.metadata.getColumn_metadata().values()) + indexNames.add(cd.getIndexName()); + } + for (ColumnDef c : cf_def.column_metadata) { - // Ensure that given idx_names and auto_generated idx_names cannot collide - CFMetaData cfm = CFMetaData.fromThrift(cf_def); - String idxName = cfm.indexName(ColumnDefinition.fromColumnDef(c)); - if (indexNames.contains(idxName)) - throw new InvalidRequestException("Duplicate index names " + idxName); - indexNames.add(idxName); - TypeParser.parse(c.validation_class); try @@ -572,11 +590,31 @@ public class ThriftValidation ByteBufferUtil.bytesToHex(c.name), cf_def.comparator_type)); } - if ((c.index_name != null) && (c.index_type == null)) - throw new ConfigurationException("index_name cannot be set without index_type"); + if (c.index_type == null) + { + if (c.index_name != null) + throw new ConfigurationException("index_name cannot be set without index_type"); + } + else + { + if (cfType == ColumnFamilyType.Super) + throw new InvalidRequestException("Secondary indexes are not supported on supercolumns"); + assert c.index_name != null; // should have a default set by now if none was provided + if (!Migration.isLegalName(c.index_name)) + throw new InvalidRequestException("Illegal index name " + c.index_name); + // check index names against this CF _and_ globally + if (indexNames.contains(c.index_name)) + throw new InvalidRequestException("Duplicate index name " + c.index_name); + indexNames.add(c.index_name); - if (cfType == ColumnFamilyType.Super && c.index_type != null) - throw new InvalidRequestException("Secondary indexes are not supported on supercolumns"); + ColumnDefinition oldCd = old == null ? null : old.getColumnDefinition(c.name); + if (oldCd != null && oldCd.getIndexType() != null) + { + assert oldCd.getIndexName() != null; + if (!oldCd.getIndexName().equals(c.index_name)) + throw new InvalidRequestException("Cannot modify index name"); + } + } } validateMinMaxCompactionThresholds(cf_def); validateMemtableSettings(cf_def); diff --git a/test/system/test_cql.py b/test/system/test_cql.py index 44c2197023..b5a0f17551 100644 --- a/test/system/test_cql.py +++ b/test/system/test_cql.py @@ -75,6 +75,10 @@ def load_sample(dbconn): CREATE COLUMNFAMILY IndexedA (KEY text PRIMARY KEY, birthdate int) WITH comparator = ascii AND default_validation = ascii; """) + dbconn.execute(""" + CREATE COLUMNFAMILY CounterCF (KEY text PRIMARY KEY, count_me counter) + WITH comparator = ascii AND default_validation = counter; + """) dbconn.execute("CREATE INDEX ON IndexedA (birthdate)") query = "UPDATE StandardString1 SET :c1 = :v1, :c2 = :v2 WHERE KEY = :key" @@ -526,7 +530,7 @@ class TestCql(ThriftTester): "creating column indexes" cursor = init() cursor.execute("USE Keyspace1") - cursor.execute("CREATE COLUMNFAMILY CreateIndex1 (KEY text PRIMARY KEY)") + cursor.execute("CREATE COLUMNFAMILY CreateIndex1 (KEY text PRIMARY KEY, items text, stuff int)") cursor.execute("CREATE INDEX namedIndex ON CreateIndex1 (items)") cursor.execute("CREATE INDEX ON CreateIndex1 (stuff)") @@ -535,10 +539,9 @@ class TestCql(ThriftTester): cfam = [i for i in ksdef.cf_defs if i.name == "CreateIndex1"][0] items = [i for i in cfam.column_metadata if i.name == "items"][0] stuff = [i for i in cfam.column_metadata if i.name == "stuff"][0] - assert items.index_name == "namedIndex", "missing index (or name)" + assert items.index_name == "namedIndex", items.index_name assert items.index_type == 0, "missing index" - assert not stuff.index_name, \ - "index_name should be unset, not %s" % stuff.index_name + assert stuff.index_name != None, "index_name should be set" assert stuff.index_type == 0, "missing index" # already indexed @@ -546,6 +549,34 @@ class TestCql(ThriftTester): cursor.execute, "CREATE INDEX ON CreateIndex1 (stuff)") + def test_drop_indexes(self): + "droping indexes on columns" + cursor = init() + cursor.execute("""CREATE KEYSPACE DropIndexTests WITH strategy_options:replication_factor = '1' + AND strategy_class = 'SimpleStrategy';""") + cursor.execute("USE DropIndexTests") + cursor.execute("CREATE COLUMNFAMILY IndexedCF (KEY text PRIMARY KEY, n text)") + cursor.execute("CREATE INDEX namedIndex ON IndexedCF (n)") + + ksdef = thrift_client.describe_keyspace("DropIndexTests") + columns = ksdef.cf_defs[0].column_metadata + + assert columns[0].index_name == "namedIndex" + assert columns[0].index_type == 0 + + # testing "DROP INDEX " + cursor.execute("DROP INDEX namedIndex") + + ksdef = thrift_client.describe_keyspace("DropIndexTests") + columns = ksdef.cf_defs[0].column_metadata + + assert columns[0].index_type == None + assert columns[0].index_name == None + + assert_raises(cql.ProgrammingError, + cursor.execute, + "DROP INDEX undefIndex") + def test_time_uuid(self): "store and retrieve time-based (type 1) uuids" cursor = init() @@ -1006,3 +1037,151 @@ class TestCql(ThriftTester): r = cursor.fetchone() assert len(r) == 1, "expected 0 results, got %d" % len(r) + + def test_alter_table_statement(self): + "test ALTER TABLE statement" + cursor = init() + cursor.execute(""" + CREATE KEYSPACE AlterTableKS WITH strategy_options:replication_factor = '1' + AND strategy_class = 'SimpleStrategy'; + """) + cursor.execute("USE AlterTableKS;") + + cursor.execute(""" + CREATE COLUMNFAMILY NewCf1 (KEY varint PRIMARY KEY) WITH default_validation = ascii; + """) + + # TODO: temporary (until this can be done with CQL). + ksdef = thrift_client.describe_keyspace("AlterTableKS") + assert len(ksdef.cf_defs) == 1, \ + "expected 1 column family total, found %d" % len(ksdef.cf_defs) + cfam = ksdef.cf_defs[0] + + assert len(cfam.column_metadata) == 0 + + # testing "add a new column" + cursor.execute("ALTER TABLE NewCf1 ADD name varchar") + + ksdef = thrift_client.describe_keyspace("AlterTableKS") + assert len(ksdef.cf_defs) == 1, \ + "expected 1 column family total, found %d" % len(ksdef.cf_defs) + columns = ksdef.cf_defs[0].column_metadata + + assert len(columns) == 1 + assert columns[0].name == 'name' + assert columns[0].validation_class == 'org.apache.cassandra.db.marshal.UTF8Type' + + # testing "alter a column type" + cursor.execute("ALTER TABLE NewCf1 ALTER name TYPE ascii") + + ksdef = thrift_client.describe_keyspace("AlterTableKS") + assert len(ksdef.cf_defs) == 1, \ + "expected 1 column family total, found %d" % len(ksdef.cf_defs) + columns = ksdef.cf_defs[0].column_metadata + + assert len(columns) == 1 + assert columns[0].name == 'name' + assert columns[0].validation_class == 'org.apache.cassandra.db.marshal.AsciiType' + + # alter column with unknown validator + assert_raises(cql.ProgrammingError, + cursor.execute, + "ALTER TABLE NewCf1 ADD name utf8") + + # testing 'drop an existing column' + cursor.execute("ALTER TABLE NewCf1 DROP name") + + ksdef = thrift_client.describe_keyspace("AlterTableKS") + assert len(ksdef.cf_defs) == 1, \ + "expected 1 column family total, found %d" % len(ksdef.cf_defs) + columns = ksdef.cf_defs[0].column_metadata + + assert len(columns) == 0 + + # add column with unknown validator + assert_raises(cql.ProgrammingError, + cursor.execute, + "ALTER TABLE NewCf1 ADD name utf8") + + # alter not existing column + assert_raises(cql.ProgrammingError, + cursor.execute, + "ALTER TABLE NewCf1 ALTER name TYPE uuid") + + # drop not existing column + assert_raises(cql.ProgrammingError, + cursor.execute, + "ALTER TABLE NewCf1 DROP name") + + def test_counter_column_support(self): + "update statement should be able to work with counter columns" + cursor = init() + + # increment counter + cursor.execute("UPDATE CounterCF SET count_me = count_me + 2 WHERE key = 'counter1'") + cursor.execute("SELECT * FROM CounterCF WHERE KEY = 'counter1'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "count_me", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == 2, \ + "unrecognized value '%s'" % r[1] + + cursor.execute("UPDATE CounterCF SET count_me = count_me + 2 WHERE key = 'counter1'") + cursor.execute("SELECT * FROM CounterCF WHERE KEY = 'counter1'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "count_me", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == 4, \ + "unrecognized value '%s'" % r[1] + + # decrement counter + cursor.execute("UPDATE CounterCF SET count_me = count_me - 4 WHERE key = 'counter1'") + cursor.execute("SELECT * FROM CounterCF WHERE KEY = 'counter1'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "count_me", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == 0, \ + "unrecognized value '%s'" % r[1] + + cursor.execute("SELECT * FROM CounterCF") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "count_me", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == 0, \ + "unrecognized value '%s'" % r[1] + + # deleting a counter column + cursor.execute("DELETE count_me FROM CounterCF WHERE KEY = 'counter1'") + cursor.execute("SELECT * FROM CounterCF") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + assert len(colnames) == 1 + + r = cursor.fetchone() + assert len(r) == 1 + + # can't mix counter and normal statements + assert_raises(cql.ProgrammingError, + cursor.execute, + "UPDATE CounterCF SET count_me = count_me + 2, x = 'a' WHERE key = 'counter1'") + + # column names must match + assert_raises(cql.ProgrammingError, + cursor.execute, + "UPDATE CounterCF SET count_me = count_not_me + 2 WHERE key = 'counter1'") diff --git a/test/system/test_thrift_server.py b/test/system/test_thrift_server.py index 4e605ddc7b..ce3bc29311 100644 --- a/test/system/test_thrift_server.py +++ b/test/system/test_thrift_server.py @@ -1421,7 +1421,7 @@ class TestMutations(ThriftTester): client.system_update_column_family(modified_cf) # Add a second indexed CF ... - birthdate_coldef = ColumnDef('birthdate', 'BytesType', IndexType.KEYS, 'birthdate_index') + birthdate_coldef = ColumnDef('birthdate', 'BytesType', IndexType.KEYS, 'birthdate2_index') age_coldef = ColumnDef('age', 'BytesType', IndexType.KEYS, 'age_index') cfdef = CfDef('Keyspace1', 'BlankCF2', column_metadata=[birthdate_coldef, age_coldef]) client.system_add_column_family(cfdef) diff --git a/test/unit/org/apache/cassandra/SchemaLoader.java b/test/unit/org/apache/cassandra/SchemaLoader.java index 0e77cc7fd2..5abfba8759 100644 --- a/test/unit/org/apache/cassandra/SchemaLoader.java +++ b/test/unit/org/apache/cassandra/SchemaLoader.java @@ -21,6 +21,7 @@ package org.apache.cassandra; import java.nio.ByteBuffer; import java.util.*; +import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.commons.lang.NotImplementedException; import com.google.common.base.Charsets; @@ -249,7 +250,7 @@ public class SchemaLoader {{ ByteBuffer cName = ByteBuffer.wrap("birthdate".getBytes(Charsets.UTF_8)); IndexType keys = withIdxType ? IndexType.KEYS : null; - put(cName, new ColumnDefinition(cName, LongType.instance, keys, null)); + put(cName, new ColumnDefinition(cName, LongType.instance, keys, ByteBufferUtil.bytesToHex(cName))); }}); } private static CFMetaData jdbcCFMD(String ksName, String cfName, AbstractType comp) diff --git a/tools/stress/bin/stressd b/tools/stress/bin/stressd new file mode 100755 index 0000000000..5cc3ff2d46 --- /dev/null +++ b/tools/stress/bin/stressd @@ -0,0 +1,87 @@ +#!/bin/sh + +# 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. + +DESC="Stress Test Daemon" + +if [ "x$CLASSPATH" = "x" ]; then + # Cassandra class files. + if [ ! -d `dirname $0`/../../../build/classes/main ]; then + echo "Unable to locate cassandra class files" >&2 + exit 1 + fi + + # Stress class files. + if [ ! -d `dirname $0`/../build/classes ]; then + echo "Unable to locate stress class files" >&2 + exit 1 + fi + + CLASSPATH=`dirname $0`/../../../build/classes/main + CLASSPATH=$CLASSPATH:`dirname $0`/../../../build/classes/thrift + CLASSPATH=$CLASSPATH:`dirname $0`/../build/classes + for jar in `dirname $0`/../../../lib/*.jar; do + CLASSPATH=$CLASSPATH:$jar + done +fi + +if [ -x $JAVA_HOME/bin/java ]; then + JAVA=$JAVA_HOME/bin/java +else + JAVA=`which java` +fi + +if [ "x$JAVA" = "x" ]; then + echo "Java executable not found (hint: set JAVA_HOME)" >&2 + exit 1 +fi + +case "$1" in + start) + echo "Starting $DESC: " + $JAVA -server -cp $CLASSPATH org.apache.cassandra.stress.StressServer $@ 1> ./stressd.out.log 2> ./stressd.err.log & + echo $! > ./stressd.pid + echo "done." + ;; + + stop) + PID=`cat ./stressd.pid 2> /dev/null` + + if [ "x$PID" = "x" ]; then + echo "$DESC is not running." + else + kill -9 $PID + rm ./stressd.pid + echo "$DESC is stopped." + fi + ;; + + status) + PID=`cat ./stressd.pid 2> /dev/null` + + if [ "x$PID" = "x" ]; then + echo "$DESC is not running." + else + echo "$DESC is running with pid $PID." + fi + ;; + + *) + echo "Usage: $0 start|stop|status [-h ]" + ;; +esac + diff --git a/tools/stress/src/org/apache/cassandra/stress/Session.java b/tools/stress/src/org/apache/cassandra/stress/Session.java index a28a671e03..e0765f8581 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Session.java +++ b/tools/stress/src/org/apache/cassandra/stress/Session.java @@ -18,6 +18,8 @@ package org.apache.cassandra.stress; import java.io.*; +import java.net.InetAddress; +import java.net.UnknownHostException; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -34,7 +36,7 @@ import org.apache.thrift.transport.TFramedTransport; import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; -public class Session +public class Session implements Serializable { // command line options public static final Options availableOptions = new Options(); @@ -74,6 +76,7 @@ public class Session availableOptions.addOption("O", "strategy-properties", true, "Replication strategy properties in the following format :,:,..."); availableOptions.addOption("W", "no-replicate-on-write",false, "Set replicate_on_write to false for counters. Only counter add with CL=ONE will work"); availableOptions.addOption("V", "average-size-values", false, "Generate column values of average rather than specific size"); + availableOptions.addOption("T", "send-to", true, "Send this as a request to the stress daemon at specified address."); } private int numKeys = 1000 * 1000; @@ -95,7 +98,7 @@ public class Session private boolean replicateOnWrite = true; private boolean ignoreErrors = false; - private PrintStream out = System.out; + private final String outFileName; private IndexType indexType = null; private Stress.Operations operation = Stress.Operations.INSERT; @@ -110,6 +113,8 @@ public class Session protected int mean; protected float sigma; + public final InetAddress sendToDaemon; + public Session(String[] arguments) throws IllegalArgumentException { float STDev = 0.1f; @@ -181,17 +186,7 @@ public class Session if (cmd.hasOption("r")) random = true; - if (cmd.hasOption("f")) - { - try - { - out = new PrintStream(new FileOutputStream(cmd.getOptionValue("f"))); - } - catch (FileNotFoundException e) - { - System.out.println(e.getMessage()); - } - } + outFileName = (cmd.hasOption("f")) ? cmd.getOptionValue("f") : null; if (cmd.hasOption("p")) port = Integer.parseInt(cmd.getOptionValue("p")); @@ -264,6 +259,17 @@ public class Session replicateOnWrite = false; averageSizeValues = cmd.hasOption("V"); + + try + { + sendToDaemon = cmd.hasOption("send-to") + ? InetAddress.getByName(cmd.getOptionValue("send-to")) + : null; + } + catch (UnknownHostException e) + { + throw new RuntimeException(e); + } } catch (ParseException e) { @@ -360,7 +366,14 @@ public class Session public PrintStream getOutputStream() { - return out; + try + { + return (outFileName == null) ? System.out : new PrintStream(new FileOutputStream(outFileName)); + } + catch (FileNotFoundException e) + { + throw new RuntimeException(e.getMessage(), e); + } } public int getProgressInterval() @@ -432,16 +445,16 @@ public class Session try { client.system_add_keyspace(keyspace); - out.println(String.format("Created keyspaces. Sleeping %ss for propagation.", nodes.length)); + System.out.println(String.format("Created keyspaces. Sleeping %ss for propagation.", nodes.length)); Thread.sleep(nodes.length * 1000); // seconds } catch (InvalidRequestException e) { - out.println("Unable to create stress keyspace: " + e.getWhy()); + System.err.println("Unable to create stress keyspace: " + e.getWhy()); } catch (Exception e) { - out.println(e.getMessage()); + System.err.println(e.getMessage()); } } diff --git a/tools/stress/src/org/apache/cassandra/stress/Stress.java b/tools/stress/src/org/apache/cassandra/stress/Stress.java index a609766c45..36f0410daa 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Stress.java +++ b/tools/stress/src/org/apache/cassandra/stress/Stress.java @@ -17,15 +17,12 @@ */ package org.apache.cassandra.stress; -import org.apache.cassandra.stress.operations.*; -import org.apache.cassandra.stress.util.Operation; -import org.apache.cassandra.thrift.Cassandra; import org.apache.commons.cli.Option; -import java.io.PrintStream; +import java.io.*; +import java.net.Socket; +import java.net.SocketException; import java.util.Random; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.SynchronousQueue; public final class Stress { @@ -36,17 +33,10 @@ public final class Stress public static Session session; public static Random randomizer = new Random(); - - /** - * Producer-Consumer model: 1 producer, N consumers - */ - private static final BlockingQueue operations = new SynchronousQueue(true); + private static volatile boolean stopped = false; public static void main(String[] arguments) throws Exception { - long latency, oldLatency; - int epoch, total, oldTotal, keyCount, oldKeyCount; - try { session = new Session(arguments); @@ -57,111 +47,49 @@ public final class Stress return; } - // creating keyspace and column families - if (session.getOperation() == Operations.INSERT || session.getOperation() == Operations.COUNTER_ADD) + PrintStream outStream = session.getOutputStream(); + + if (session.sendToDaemon != null) { - session.createKeySpaces(); - } + Socket socket = new Socket(session.sendToDaemon, 2159); - int threadCount = session.getThreads(); - Thread[] consumers = new Thread[threadCount]; - PrintStream out = session.getOutputStream(); + ObjectOutputStream out = new ObjectOutputStream(socket.getOutputStream()); + BufferedReader inp = new BufferedReader(new InputStreamReader(socket.getInputStream())); - out.println("total,interval_op_rate,interval_key_rate,avg_latency,elapsed_time"); + Runtime.getRuntime().addShutdownHook(new ShutDown(socket, out)); - int itemsPerThread = session.getKeysPerThread(); - int modulo = session.getNumKeys() % threadCount; + out.writeObject(session); - // creating required type of the threads for the test - for (int i = 0; i < threadCount; i++) - { - if (i == threadCount - 1) - itemsPerThread += modulo; // last one is going to handle N + modulo items + String line; - consumers[i] = new Consumer(itemsPerThread); - } - - new Producer().start(); - - // starting worker threads - for (int i = 0; i < threadCount; i++) - { - consumers[i].start(); - } - - // initialization of the values - boolean terminate = false; - latency = 0; - epoch = total = keyCount = 0; - - int interval = session.getProgressInterval(); - int epochIntervals = session.getProgressInterval() * 10; - long testStartTime = System.currentTimeMillis(); - - while (!terminate) - { - Thread.sleep(100); - - int alive = 0; - for (Thread thread : consumers) - if (thread.isAlive()) alive++; - - if (alive == 0) - terminate = true; - - epoch++; - - if (terminate || epoch > epochIntervals) + try { - epoch = 0; + while (!socket.isClosed() && (line = inp.readLine()) != null) + { + if (line.equals("END")) + { + out.writeInt(1); + break; + } - oldTotal = total; - oldLatency = latency; - oldKeyCount = keyCount; - - total = session.operations.get(); - keyCount = session.keys.get(); - latency = session.latency.get(); - - int opDelta = total - oldTotal; - int keyDelta = keyCount - oldKeyCount; - double latencyDelta = latency - oldLatency; - - long currentTimeInSeconds = (System.currentTimeMillis() - testStartTime) / 1000; - String formattedDelta = (opDelta > 0) ? Double.toString(latencyDelta / (opDelta * 1000)) : "NaN"; - - out.println(String.format("%d,%d,%d,%s,%d", total, opDelta / interval, keyDelta / interval, formattedDelta, currentTimeInSeconds)); + outStream.println(line); + } + } + catch (SocketException e) + { + if (!stopped) + e.printStackTrace(); } - } - } - private static Operation createOperation(int index) - { - switch (session.getOperation()) + out.close(); + inp.close(); + + socket.close(); + } + else { - case READ: - return new Reader(index); - - case COUNTER_GET: - return new CounterGetter(index); - - case INSERT: - return new Inserter(index); - - case COUNTER_ADD: - return new CounterAdder(index); - - case RANGE_SLICE: - return new RangeSlicer(index); - - case INDEXED_RANGE_SLICE: - return new IndexedRangeSlicer(index); - - case MULTI_GET: - return new MultiGetter(index); + new StressAction(session, outStream).run(); } - - throw new UnsupportedOperationException(); } /** @@ -180,55 +108,34 @@ public final class Stress } } - /** - * Produces exactly N items (awaits each to be consumed) - */ - private static class Producer extends Thread + private static class ShutDown extends Thread { + private final Socket socket; + private final ObjectOutputStream out; + + public ShutDown(Socket socket, ObjectOutputStream out) + { + this.out = out; + this.socket = socket; + } + public void run() { - for (int i = 0; i < session.getNumKeys(); i++) + try { - try + if (!socket.isClosed()) { - operations.put(createOperation(i % session.getNumDifferentKeys())); - } - catch (InterruptedException e) - { - System.err.println("Producer error - " + e.getMessage()); - return; + System.out.println("Control-C caught. Canceling running action and shutting down..."); + + out.writeInt(1); + out.close(); + + stopped = true; } } - } - } - - /** - * Each consumes exactly N items from queue - */ - private static class Consumer extends Thread - { - private final int items; - - public Consumer(int toConsume) - { - items = toConsume; - } - - public void run() - { - Cassandra.Client client = session.getClient(); - - for (int i = 0; i < items; i++) + catch (IOException e) { - try - { - operations.take().run(client); // running job - } - catch (Exception e) - { - System.err.println(e.getMessage()); - System.exit(-1); - } + e.printStackTrace(); } } } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java new file mode 100644 index 0000000000..571d498f17 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java @@ -0,0 +1,256 @@ +/** + * 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.stress; + +import java.io.PrintStream; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.SynchronousQueue; + +import org.apache.cassandra.stress.operations.*; +import org.apache.cassandra.stress.util.Operation; +import org.apache.cassandra.thrift.Cassandra; + +public class StressAction extends Thread +{ + /** + * Producer-Consumer model: 1 producer, N consumers + */ + private final BlockingQueue operations = new SynchronousQueue(true); + + private final Session client; + private final PrintStream output; + + private volatile boolean stop = false; + + public StressAction(Session session, PrintStream out) + { + client = session; + output = out; + } + + public void run() + { + long latency, oldLatency; + int epoch, total, oldTotal, keyCount, oldKeyCount; + + // creating keyspace and column families + if (client.getOperation() == Stress.Operations.INSERT || client.getOperation() == Stress.Operations.COUNTER_ADD) + client.createKeySpaces(); + + int threadCount = client.getThreads(); + Consumer[] consumers = new Consumer[threadCount]; + + output.println("total,interval_op_rate,interval_key_rate,avg_latency,elapsed_time"); + + int itemsPerThread = client.getKeysPerThread(); + int modulo = client.getNumKeys() % threadCount; + + // creating required type of the threads for the test + for (int i = 0; i < threadCount; i++) { + if (i == threadCount - 1) + itemsPerThread += modulo; // last one is going to handle N + modulo items + + consumers[i] = new Consumer(itemsPerThread); + } + + Producer producer = new Producer(); + producer.start(); + + // starting worker threads + for (int i = 0; i < threadCount; i++) + consumers[i].start(); + + // initialization of the values + boolean terminate = false; + latency = 0; + epoch = total = keyCount = 0; + + int interval = client.getProgressInterval(); + int epochIntervals = client.getProgressInterval() * 10; + long testStartTime = System.currentTimeMillis(); + + while (!terminate) + { + if (stop) + { + producer.stopProducer(); + + for (Consumer consumer : consumers) + consumer.stopConsume(); + + break; + } + + try + { + Thread.sleep(100); + } + catch (InterruptedException e) + { + throw new RuntimeException(e.getMessage(), e); + } + + int alive = 0; + for (Thread thread : consumers) + if (thread.isAlive()) alive++; + + if (alive == 0) + terminate = true; + + epoch++; + + if (terminate || epoch > epochIntervals) + { + epoch = 0; + + oldTotal = total; + oldLatency = latency; + oldKeyCount = keyCount; + + total = client.operations.get(); + keyCount = client.keys.get(); + latency = client.latency.get(); + + int opDelta = total - oldTotal; + int keyDelta = keyCount - oldKeyCount; + double latencyDelta = latency - oldLatency; + + long currentTimeInSeconds = (System.currentTimeMillis() - testStartTime) / 1000; + String formattedDelta = (opDelta > 0) ? Double.toString(latencyDelta / (opDelta * 1000)) : "NaN"; + + output.println(String.format("%d,%d,%d,%s,%d", total, opDelta / interval, keyDelta / interval, formattedDelta, currentTimeInSeconds)); + } + } + + // marking an end of the output to the client + output.println("END"); + } + + /** + * Produces exactly N items (awaits each to be consumed) + */ + private class Producer extends Thread + { + private volatile boolean stop = false; + + public void run() + { + for (int i = 0; i < client.getNumKeys(); i++) + { + if (stop) + break; + + try + { + operations.put(createOperation(i % client.getNumDifferentKeys())); + } + catch (InterruptedException e) + { + System.err.println("Producer error - " + e.getMessage()); + return; + } + } + } + + public void stopProducer() + { + stop = true; + } + } + + /** + * Each consumes exactly N items from queue + */ + private class Consumer extends Thread + { + private final int items; + private volatile boolean stop = false; + + public Consumer(int toConsume) + { + items = toConsume; + } + + public void run() + { + Cassandra.Client connection = client.getClient(); + + for (int i = 0; i < items; i++) + { + if (stop) + break; + + try + { + operations.take().run(connection); // running job + } + catch (Exception e) + { + if (output == null) + { + System.err.println(e.getMessage()); + System.exit(-1); + } + + + output.println(e.getMessage()); + break; + } + } + } + + public void stopConsume() + { + stop = true; + } + } + + private Operation createOperation(int index) + { + switch (client.getOperation()) + { + case READ: + return new Reader(client, index); + + case COUNTER_GET: + return new CounterGetter(client, index); + + case INSERT: + return new Inserter(client, index); + + case COUNTER_ADD: + return new CounterAdder(client, index); + + case RANGE_SLICE: + return new RangeSlicer(client, index); + + case INDEXED_RANGE_SLICE: + return new IndexedRangeSlicer(client, index); + + case MULTI_GET: + return new MultiGetter(client, index); + } + + throw new UnsupportedOperationException(); + } + + public void stopAction() + { + stop = true; + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/StressServer.java b/tools/stress/src/org/apache/cassandra/stress/StressServer.java new file mode 100644 index 0000000000..6600dfd444 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/StressServer.java @@ -0,0 +1,71 @@ +/** + * 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.stress; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; + +import org.apache.cassandra.stress.server.StressThread; +import org.apache.commons.cli.*; + +public class StressServer +{ + private static final Options availableOptions = new Options(); + + static + { + availableOptions.addOption("h", "host", true, "Host to listen for connections."); + } + + public static void main(String[] args) throws Exception + { + ServerSocket serverSocket = null; + CommandLineParser parser = new PosixParser(); + + InetAddress address = InetAddress.getByName("127.0.0.1"); + + try + { + CommandLine cmd = parser.parse(availableOptions, args); + + if (cmd.hasOption("h")) + { + address = InetAddress.getByName(cmd.getOptionValue("h")); + } + } + catch (ParseException e) + { + System.err.printf("Usage: ./bin/stressd start|stop|status [-h ]"); + System.exit(1); + } + + try + { + serverSocket = new ServerSocket(2159, 0, address); + } + catch (IOException e) + { + System.err.printf("Could not listen on port: %s:2159.%n", address.getHostAddress()); + System.exit(1); + } + + for (;;) + new StressThread(serverSocket.accept()).start(); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/CounterAdder.java index adbc2d2025..0c80f0a71c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CounterAdder.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.db.ColumnFamilyType; import org.apache.cassandra.thrift.*; @@ -31,9 +32,9 @@ import java.util.Map; public class CounterAdder extends Operation { - public CounterAdder(int index) + public CounterAdder(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/CounterGetter.java index 09dcb32954..4c74aee414 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CounterGetter.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.db.ColumnFamilyType; import org.apache.cassandra.thrift.*; @@ -27,9 +28,9 @@ import java.util.List; public class CounterGetter extends Operation { - public CounterGetter(int index) + public CounterGetter(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/IndexedRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/IndexedRangeSlicer.java index f3957c28e8..c117862428 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/IndexedRangeSlicer.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/IndexedRangeSlicer.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.thrift.*; import org.apache.cassandra.utils.ByteBufferUtil; @@ -30,9 +31,9 @@ public class IndexedRangeSlicer extends Operation { private static List values = null; - public IndexedRangeSlicer(int index) + public IndexedRangeSlicer(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/Inserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/Inserter.java index 89f1203099..45d33cba28 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/Inserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/Inserter.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.db.ColumnFamilyType; import org.apache.cassandra.thrift.*; @@ -33,9 +34,9 @@ public class Inserter extends Operation { private static List values; - public Inserter(int index) + public Inserter(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/MultiGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/MultiGetter.java index 096798db59..c50dd1bf80 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/MultiGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/MultiGetter.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.db.ColumnFamilyType; import org.apache.cassandra.thrift.*; @@ -31,9 +32,9 @@ import java.util.Map; public class MultiGetter extends Operation { - public MultiGetter(int index) + public MultiGetter(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/RangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/RangeSlicer.java index 46ec4a0fcf..308eefe356 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/RangeSlicer.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/RangeSlicer.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.db.ColumnFamilyType; import org.apache.cassandra.thrift.*; @@ -30,9 +31,9 @@ import java.util.List; public class RangeSlicer extends Operation { - public RangeSlicer(int index) + public RangeSlicer(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java b/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java index cf88fc6c9d..6dcd4cd6fb 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.stress.operations; +import org.apache.cassandra.stress.Session; import org.apache.cassandra.stress.util.Operation; import org.apache.cassandra.db.ColumnFamilyType; import org.apache.cassandra.thrift.*; @@ -29,9 +30,9 @@ import static com.google.common.base.Charsets.UTF_8; public class Reader extends Operation { - public Reader(int index) + public Reader(Session client, int index) { - super(index); + super(client, index); } public void run(Cassandra.Client client) throws IOException diff --git a/tools/stress/src/org/apache/cassandra/stress/server/StressThread.java b/tools/stress/src/org/apache/cassandra/stress/server/StressThread.java new file mode 100644 index 0000000000..3d6fb0deba --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/server/StressThread.java @@ -0,0 +1,79 @@ +/** + * 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.stress.server; + +import org.apache.cassandra.stress.Session; +import org.apache.cassandra.stress.StressAction; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.PrintStream; +import java.net.Socket; +import java.net.SocketException; +import java.net.SocketTimeoutException; + +public class StressThread extends Thread +{ + private final Socket socket; + + public StressThread(Socket client) + { + this.socket = client; + } + + public void run() + { + try + { + ObjectInputStream in = new ObjectInputStream(socket.getInputStream()); + PrintStream out = new PrintStream(socket.getOutputStream()); + + StressAction action = new StressAction((Session) in.readObject(), out); + action.start(); + + while (action.isAlive()) + { + try + { + if (in.readInt() == 1) + { + action.stopAction(); + break; + } + } + catch (Exception e) + { + // continue without problem + } + } + + out.close(); + in.close(); + socket.close(); + } + catch (IOException e) + { + throw new RuntimeException(e.getMessage(), e); + } + catch (Exception e) + { + e.printStackTrace(); + } + } + +} diff --git a/tools/stress/src/org/apache/cassandra/stress/util/Operation.java b/tools/stress/src/org/apache/cassandra/stress/util/Operation.java index 5ec6f5b3ec..10b97ad6cf 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/Operation.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/Operation.java @@ -46,6 +46,12 @@ public abstract class Operation session = Stress.session; } + public Operation(Session client, int idx) + { + index = idx; + session = client; + } + /** * Run operation * @param client Cassandra Thrift client connection @@ -101,18 +107,18 @@ public abstract class Operation * key generator using Gauss or Random algorithm * @return byte[] representation of the key string */ - protected static byte[] generateKey() + protected byte[] generateKey() { - return (Stress.session.useRandomGenerator()) ? generateRandomKey() : generateGaussKey(); + return (session.useRandomGenerator()) ? generateRandomKey() : generateGaussKey(); } /** * Random key generator * @return byte[] representation of the key string */ - private static byte[] generateRandomKey() + private byte[] generateRandomKey() { - String format = "%0" + Stress.session.getTotalKeysLength() + "d"; + String format = "%0" + session.getTotalKeysLength() + "d"; return String.format(format, Stress.randomizer.nextInt(Stress.session.getNumDifferentKeys() - 1)).getBytes(UTF_8); } @@ -120,9 +126,8 @@ public abstract class Operation * Gauss key generator * @return byte[] representation of the key string */ - private static byte[] generateGaussKey() + private byte[] generateGaussKey() { - Session session = Stress.session; String format = "%0" + session.getTotalKeysLength() + "d"; for (;;)