natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

merge from 0.8 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1132406 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-06-05 13:43:22 +00:00
parent e1941e2128 a7bc6ee07c
commit f5f96c178d
48 changed files with 1643 additions and 597 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGES.txt
View File

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

9
build.xml
View File

@ -64,7 +64,7 @@
<property name="test.long.src" value="${test.dir}/long"/>
<property name="test.distributed.src" value="${test.dir}/distributed"/>
<property name="dist.dir" value="${build.dir}/dist"/>
<property name="cql.driver.version" value="1.0.2" />
<property name="cql.driver.version" value="1.0.3" />
<condition property="version" value="${base.version}">
<isset property="release"/>
</condition>
@ -693,7 +693,6 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
destdir="${build.classes.main}" includeantruntime="false">
<src path="${build.src.java}"/>
<src path="${build.src.gen-java}"/>
<src path="${build.src.driver}" />
<classpath refid="cassandra.classpath"/>
</javac>
<javac debug="true" debuglevel="${debuglevel}"
@ -972,7 +971,10 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
debug="true"
debuglevel="${debuglevel}"
destdir="${test.classes}">
<classpath refid="cassandra.classpath"/>
<classpath>
<path refid="cassandra.classpath"/>
<pathelement location="${build.classes.cql}"/>
</classpath>
<src path="${test.unit.src}"/>
<src path="${test.long.src}"/>
<src path="${test.src.driver}"/>
@ -1024,6 +1026,7 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<classpath>
<path refid="cassandra.classpath" />
<pathelement location="${test.classes}"/>
<pathelement location="${build.classes.cql}"/>
<path refid="cobertura.classpath"/>
<pathelement location="${test.conf}"/>
<fileset dir="${test.lib}">

6
contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
View File

@ -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<org.apache.cassandra.hadoop.avro.Column> columns = new ArrayList<org.apache.cassandra.hadoop.avro.Column>();
ArrayList<org.apache.cassandra.thrift.Column> columns = new ArrayList<org.apache.cassandra.thrift.Column>();
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);

2
debian/changelog vendored
View File

@ -2,7 +2,7 @@ cassandra (0.8.0) unstable; urgency=low
* New release
-- Eric Evans <eevans@apache.org> Mon, 23 May 2011 15:59:48 -0500
-- Eric Evans <eevans@apache.org> Mon, 30 May 2011 12:58:23 -0500
cassandra (0.8.0~rc1) unstable; urgency=low

20
doc/cql/CQL.textile
View File

@ -69,6 +69,17 @@ SELECT ... WHERE <CLAUSE> [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 <columnFamily> ADD <column> <validator>;
ALTER TABLE <columnFamily> ALTER <column> TYPE <validator>;
ALTER TABLE <columnFamily> DROP <column>;
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_family> (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 <INDEX_NAME>
A @DROP INDEX@ statement is used to drop an existing secondary index.
DROP INDEX <INDEX_NAME> statement will search all ColumnFamilies in the current Keyspace for specified index and delete it if found.
h2. DROP
_Synopsis:_

8
drivers/java/src/org/apache/cassandra/cql/jdbc/CResultSet.java
View File

@ -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<TypedColumn> values = new ArrayList<TypedColumn>();
@ -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<Column> cols = row.getColumns();
for (Column col : cols)
{

218
drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraConnection.java
View File

@ -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 <T>
* @param arg0
@ -140,17 +144,19 @@ class CassandraConnection implements Connection
*/
public <T> T unwrap(Class<T> 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<String, Class<?>> 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<String, Class<?>> arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
throw new SQLFeatureNotSupportedException("the Cassandra Implementation does not support this method");
}
}

1
drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraResultSet.java
View File

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

10
drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java
View File

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

57
drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java
View File

@ -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<String> 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<String> 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<String> keys, String... cols) throws SQLException
{
int actualRows = 0;
assert rs != null;
Iterator<String> 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.

18
drivers/py/cql/marshal.py
View File

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

49
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -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<String, String>(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()
{

5
src/java/org/apache/cassandra/config/KSMetaData.java
View File

@ -68,7 +68,10 @@ public final class KSMetaData
public static Map<String, String> forwardsCompatibleOptions(KsDef ks_def)
{
Map<String, String> options = new HashMap<String, String>(ks_def.strategy_options);
Map<String, String> options;
options = ks_def.strategy_options == null
? new HashMap<String, String>()
: new HashMap<String, String>(ks_def.strategy_options);
maybeAddReplicationFactor(options, ks_def.strategy_class, ks_def.isSetReplication_factor() ? ks_def.replication_factor : null);
return options;
}

41
src/java/org/apache/cassandra/cql/AbstractModification.java
View File

@ -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<RowMutation> prepareRowMutations(String keyspace, ClientState clientState)
throws InvalidRequestException;
public abstract List<IMutation> 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<RowMutation> 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<Term> getKeys();
public abstract List<IMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp)
throws org.apache.cassandra.thrift.InvalidRequestException;
}

122
src/java/org/apache/cassandra/cql/AlterTableStatement.java Normal file
View File

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

76
src/java/org/apache/cassandra/cql/BatchStatement.java
View File

@ -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 <code>BATCH</code> statement parsed from a CQL query.
*
@ -85,41 +76,12 @@ public class BatchStatement
return timeToLive;
}
public List<RowMutation> getMutations(String keyspace, ClientState clientState) throws InvalidRequestException
public List<IMutation> getMutations(String keyspace, ClientState clientState) throws InvalidRequestException
{
// To avoid unnecessary authorizations.
List<String> seenColumnFamilies = new ArrayList<String>();
List<IMutation> batch = new LinkedList<IMutation>();
List<RowMutation> batch = new LinkedList<RowMutation>();
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<ColumnFamily> 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<String> 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);

50
src/java/org/apache/cassandra/cql/Cql.g
View File

@ -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 <KEYSPACE>;
@ -288,12 +292,12 @@ batchStatementObjective returns [AbstractModification statement]
updateStatement returns [UpdateStatement expr]
: {
Attributes attrs = new Attributes();
Map<Term, Term> columns = new HashMap<Term, Term>();
Map<Term, Operation> columns = new HashMap<Term, Operation>();
List<Term> 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 <CF>.<COLUMN_OR_INDEX_NAME>
* DROP INDEX <INDEX_NAME>
*/
dropIndexStatement returns [DropIndexStatement expr]
:
K_DROP K_INDEX index=( IDENT | STRING_LITERAL | INTEGER ) endStmnt
{ $expr = new DropIndexStatement($index.text); }
;
/** DROP KEYSPACE <KSP>; */
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 <CF>; */
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<Term, Term> columns]
: key=term '=' value=term { columns.put(key, value); }
;
termPairWithOperation[Map<Term, Operation> 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');

3
src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
View File

@ -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<String, String> comparators = new HashMap<String, String>();
public static final Map<String, String> comparators = new HashMap<String, String>();
private static final Set<String> keywords = new HashSet<String>();
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);

10
src/java/org/apache/cassandra/cql/DeleteStatement.java
View File

@ -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<RowMutation> prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException
public List<IMutation> prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException
{
return prepareRowMutations(keyspace, clientState, null);
}
/** {@inheritDoc} */
public List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
public List<IMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
{
clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
AbstractType<?> keyType = DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator();
List<RowMutation> rowMutations = new ArrayList<RowMutation>();
List<IMutation> rowMutations = new ArrayList<IMutation>();
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

58
src/java/org/apache/cassandra/cql/Operation.java Normal file
View File

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

134
src/java/org/apache/cassandra/cql/QueryProcessor.java
View File

@ -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<ReadCommand> commands = new ArrayList<ReadCommand>();
// ...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<org.apache.cassandra.db.Row> 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<RowMutation> rowMutations = new ArrayList<RowMutation>();
List<IMutation> rowMutations = new ArrayList<IMutation>();
List<String> cfamsSeen = new ArrayList<String>();
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<org.apache.cassandra.db.Row> rows;
@ -640,7 +640,7 @@ public class QueryProcessor
{
KsDef ksd = new KsDef(create.getName(),
create.getStrategyClass(),
Collections.<org.apache.cassandra.thrift.CfDef>emptyList())
Collections.<CfDef>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;

4
src/java/org/apache/cassandra/cql/StatementType.java
View File

@ -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<StatementType> topLevel = EnumSet.of(USE, CREATE_KEYSPACE, DROP_KEYSPACE);

112
src/java/org/apache/cassandra/cql/UpdateStatement.java
View File

@ -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<Term, Term> columns;
private Map<Term, Operation> columns;
private List<Term> columnNames, columnValues;
private List<Term> keys;
@ -57,7 +60,7 @@ public class UpdateStatement extends AbstractModification
* @param attrs additional attributes for statement (CL, timestamp, timeToLive)
*/
public UpdateStatement(String columnFamily,
Map<Term, Term> columns,
Map<Term, Operation> columns,
List<Term> keys,
Attributes attrs)
{
@ -113,17 +116,28 @@ public class UpdateStatement extends AbstractModification
}
/** {@inheritDoc} */
public List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException
public List<IMutation> prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException
{
return prepareRowMutations(keyspace, clientState, null);
}
/** {@inheritDoc} */
public List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
public List<IMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
{
List<String> cfamsSeen = new ArrayList<String>();
CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false);
boolean hasCommutativeOperation = false;
for (Map.Entry<Term, Operation> 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<RowMutation> rowMutations = new LinkedList<RowMutation>();
List<IMutation> rowMutations = new LinkedList<IMutation>();
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<Term, Term> column : getColumns().entrySet())
// if true we need to wrap RowMutation into CounterMutation
boolean hasCounterColumn = false;
RowMutation rm = new RowMutation(keyspace, key);
for (Map.Entry<Term, Operation> 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 + <long> 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<Term, Term> getColumns() throws InvalidRequestException
public Map<Term, Operation> 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<Term, Term>();
columns = new HashMap<Term, Operation>();
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;
}

47
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -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<SSTableReader> truncatedSSTables = new ArrayList<SSTableReader>();
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.

25
src/java/org/apache/cassandra/db/CompactionManager.java
View File

@ -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<SSTableReader> truncatedSSTables = new ArrayList<SSTableReader>();
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()

39
src/java/org/apache/cassandra/db/DataTracker.java
View File

@ -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<SSTableReader> sstables;
public final Set<SSTableReader> compacting;
public View(Memtable memtable, Set<Memtable> pendingFlush, Set<SSTableReader> sstables, Set<SSTableReader> compacting)
View(Memtable memtable, Set<Memtable> pendingFlush, Set<SSTableReader> sstables, Set<SSTableReader> 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<Memtable> newPending = new HashSet<Memtable>(memtablesPendingFlush);
newPending.add(memtable);
Set<Memtable> newPending = ImmutableSet.<Memtable>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<Memtable> newPendings = new HashSet<Memtable>(memtablesPendingFlush);
Set<SSTableReader> newSSTables = new HashSet<SSTableReader>(sstables);
newPendings.remove(flushedMemtable);
newSSTables.add(newSSTable);
return new View(memtable, newPendings, newSSTables, compacting);
Set<Memtable> newPending = ImmutableSet.copyOf(Sets.difference(memtablesPendingFlush, Collections.singleton(flushedMemtable)));
Set<SSTableReader> newSSTables = ImmutableSet.<SSTableReader>builder().addAll(sstables).add(newSSTable).build();
return new View(memtable, newPending, newSSTables, compacting);
}
public View replace(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements)
{
Set<SSTableReader> sstablesNew = new HashSet<SSTableReader>(sstables);
Iterables.addAll(sstablesNew, replacements);
sstablesNew.removeAll(oldSSTables);
return new View(memtable, memtablesPendingFlush, sstablesNew, compacting);
Sets.SetView<SSTableReader> remaining = Sets.difference(sstables, ImmutableSet.copyOf(oldSSTables));
Set<SSTableReader> newSSTables = ImmutableSet.<SSTableReader>builder().addAll(remaining).addAll(replacements).build();
return new View(memtable, memtablesPendingFlush, newSSTables, compacting);
}
public View markCompacting(Collection<SSTableReader> tomark)
{
Set<SSTableReader> compactingNew = new HashSet<SSTableReader>(compacting);
compactingNew.addAll(tomark);
Set<SSTableReader> compactingNew = ImmutableSet.<SSTableReader>builder().addAll(sstables).addAll(tomark).build();
return new View(memtable, memtablesPendingFlush, sstables, compactingNew);
}
public View unmarkCompacting(Collection<SSTableReader> tounmark)
{
Set<SSTableReader> compactingNew = new HashSet<SSTableReader>(compacting);
compactingNew.removeAll(tounmark);
Set<SSTableReader> compactingNew = ImmutableSet.copyOf(Sets.difference(compacting, ImmutableSet.copyOf(tounmark)));
return new View(memtable, memtablesPendingFlush, sstables, compactingNew);
}
}

17
src/java/org/apache/cassandra/db/Table.java
View File

@ -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()
{

4
src/java/org/apache/cassandra/db/TruncateVerbHandler.java
View File

@ -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)
{

11
src/java/org/apache/cassandra/net/IncomingTcpConnection.java
View File

@ -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

13
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -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<CFMetaData> cfDefs = new ArrayList<CFMetaData>(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();

62
src/java/org/apache/cassandra/thrift/ThriftValidation.java
View File

@ -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<String> indexNames = new HashSet<String>();
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);

187
test/system/test_cql.py
View File

@ -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 <INDEX_NAME>"
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'")

2
test/system/test_thrift_server.py
View File

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

3
test/unit/org/apache/cassandra/SchemaLoader.java
View File

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

87
tools/stress/bin/stressd Executable file
View File

@ -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 <host>]"
;;
esac

47
tools/stress/src/org/apache/cassandra/stress/Session.java
View File

@ -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 <dc_name>:<num>,<dc_name>:<num>,...");
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());
}
}

203
tools/stress/src/org/apache/cassandra/stress/Stress.java
View File

@ -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<Operation> operations = new SynchronousQueue<Operation>(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();
}
}
}

256
tools/stress/src/org/apache/cassandra/stress/StressAction.java Normal file
View File

@ -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<Operation> operations = new SynchronousQueue<Operation>(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;
}
}

71
tools/stress/src/org/apache/cassandra/stress/StressServer.java Normal file
View File

@ -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 <host>]");
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();
}
}

5
tools/stress/src/org/apache/cassandra/stress/operations/CounterAdder.java
View File

@ -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

5
tools/stress/src/org/apache/cassandra/stress/operations/CounterGetter.java
View File

@ -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

5
tools/stress/src/org/apache/cassandra/stress/operations/IndexedRangeSlicer.java
View File

@ -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<ByteBuffer> 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

5
tools/stress/src/org/apache/cassandra/stress/operations/Inserter.java
View File

@ -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<ByteBuffer> values;
public Inserter(int index)
public Inserter(Session client, int index)
{
super(index);
super(client, index);
}
public void run(Cassandra.Client client) throws IOException

5
tools/stress/src/org/apache/cassandra/stress/operations/MultiGetter.java
View File

@ -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

5
tools/stress/src/org/apache/cassandra/stress/operations/RangeSlicer.java
View File

@ -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

5
tools/stress/src/org/apache/cassandra/stress/operations/Reader.java
View File

@ -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

79
tools/stress/src/org/apache/cassandra/stress/server/StressThread.java Normal file
View File

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

17
tools/stress/src/org/apache/cassandra/stress/util/Operation.java
View File

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