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@1100900 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-05-09 07:05:55 +00:00
parent 5798effe33
commit 835669aee8
48 changed files with 2245 additions and 3277 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGES.txt
View File

@ -12,6 +12,12 @@
* fix returning null column values in the python cql driver (CASSANDRA-2593)
* snapshot_before_compaction directory name fix (CASSANDRA-2598)
* fix CQL treatment of > and < operators in range slices (CASSANDRA-2592)
* fix potential double-application of counter updates on commitlog replay
(CASSANDRA-2419)
* JDBC CQL driver exposes getColumn for access to timestamp
* JDBC ResultSetMetadata properties added to AbstractType
0.8.0-beta2
* fix NPE compacting index CFs (CASSANDRA-2528)

4
NEWS.txt
View File

@ -6,6 +6,10 @@ Upgrading
- Upgrading from version 0.7.1 or later can be done with a rolling
restart, one node at a time. You do not need to bring down the
whole cluster.
- Running nodetool drain before shutting down the 0.7 node is
recommended but not required. (Skipping this will result in
replay of entire commitlog, so it will take longer to restart but
is otherwise harmless.)
- Avro record classes used in map/reduce and Hadoop streaming code have
moved from org.apache.cassandra.avro to org.apache.cassandra.hadoop.avro,
applications using these classes will need to be updated accordingly.

740
drivers/java/src/org/apache/cassandra/cql/jdbc/AbstractResultSet.java Normal file
View File

@ -0,0 +1,740 @@
package org.apache.cassandra.cql.jdbc;
import java.io.InputStream;
import java.io.Reader;
import java.math.BigDecimal;
import java.net.URL;
import java.sql.*;
/** a class to hold all the unimplemented crap */
class AbstractResultSet
{
public boolean absolute(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void afterLast() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void beforeFirst() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void cancelRowUpdates() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void clearWarnings() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void deleteRow() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int findColumn(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean first() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Array getArray(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Array getArray(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public InputStream getAsciiStream(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public InputStream getAsciiStream(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public InputStream getBinaryStream(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public InputStream getBinaryStream(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Blob getBlob(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Blob getBlob(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public byte getByte(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public byte getByte(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Reader getCharacterStream(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Reader getCharacterStream(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Clob getClob(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Clob getClob(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int getConcurrency() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public String getCursorName() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int getFetchDirection() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int getFetchSize() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int getHoldability() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Ref getRef(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Ref getRef(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public RowId getRowId(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public RowId getRowId(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public SQLXML getSQLXML(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public SQLXML getSQLXML(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public short getShort(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public short getShort(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Statement getStatement() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public URL getURL(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public URL getURL(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public InputStream getUnicodeStream(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public InputStream getUnicodeStream(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public SQLWarning getWarnings() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void insertRow() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void moveToCurrentRow() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void moveToInsertRow() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean previous() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void refreshRow() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean relative(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean rowDeleted() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean rowInserted() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean rowUpdated() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void setFetchDirection(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void setFetchSize(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Reader getNCharacterStream(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Reader getNCharacterStream(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public NClob getNClob(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public NClob getNClob(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public String getNString(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public String getNString(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public BigDecimal getBigDecimal(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public BigDecimal getBigDecimal(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public BigDecimal getBigDecimal(int arg0, int arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public BigDecimal getBigDecimal(String arg0, int arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
//
// all the update methods are unsupported, requires a separate statement in Cassandra
//
public void updateArray(int arg0, Array arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateArray(String arg0, Array arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateAsciiStream(int arg0, InputStream arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateAsciiStream(String arg0, InputStream arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateAsciiStream(int arg0, InputStream arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateAsciiStream(String arg0, InputStream arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateAsciiStream(int arg0, InputStream arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateAsciiStream(String arg0, InputStream arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBigDecimal(int arg0, BigDecimal arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBigDecimal(String arg0, BigDecimal arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBinaryStream(int arg0, InputStream arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBinaryStream(String arg0, InputStream arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBinaryStream(int arg0, InputStream arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBinaryStream(String arg0, InputStream arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBinaryStream(int arg0, InputStream arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBinaryStream(String arg0, InputStream arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBlob(int arg0, Blob arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBlob(String arg0, Blob arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBlob(int arg0, InputStream arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBlob(String arg0, InputStream arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBlob(int arg0, InputStream arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBlob(String arg0, InputStream arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBoolean(int arg0, boolean arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBoolean(String arg0, boolean arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateByte(int arg0, byte arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateByte(String arg0, byte arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBytes(int arg0, byte[] arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateBytes(String arg0, byte[] arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateCharacterStream(int arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateCharacterStream(String arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateCharacterStream(int arg0, Reader arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateCharacterStream(String arg0, Reader arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateCharacterStream(int arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateCharacterStream(String arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateClob(int arg0, Clob arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateClob(String arg0, Clob arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateClob(int arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateClob(String arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateClob(int arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateClob(String arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateDate(int arg0, Date arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateDate(String arg0, Date arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateDouble(int arg0, double arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateDouble(String arg0, double arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateFloat(int arg0, float arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateFloat(String arg0, float arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateInt(int arg0, int arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateInt(String arg0, int arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateLong(int arg0, long arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateLong(String arg0, long arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNCharacterStream(int arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNCharacterStream(String arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNCharacterStream(int arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNCharacterStream(String arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNClob(int arg0, NClob arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNClob(String arg0, NClob arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNClob(int arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNClob(String arg0, Reader arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNClob(int arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNClob(String arg0, Reader arg1, long arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNString(int arg0, String arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNString(String arg0, String arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNull(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateNull(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateObject(int arg0, Object arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateObject(String arg0, Object arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateObject(int arg0, Object arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateObject(String arg0, Object arg1, int arg2) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateRef(int arg0, Ref arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateRef(String arg0, Ref arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateRow() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateRowId(int arg0, RowId arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateRowId(String arg0, RowId arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateSQLXML(int arg0, SQLXML arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateSQLXML(String arg0, SQLXML arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateShort(int arg0, short arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateShort(String arg0, short arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateString(int arg0, String arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateString(String arg0, String arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateTime(int arg0, Time arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateTime(String arg0, Time arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateTimestamp(int arg0, Timestamp arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public void updateTimestamp(String arg0, Timestamp arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
}

550
drivers/java/src/org/apache/cassandra/cql/jdbc/CResultSet.java Normal file
View File

@ -0,0 +1,550 @@
/*
*
* 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.jdbc;
import java.io.InputStream;
import java.io.Reader;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.net.URL;
import java.nio.ByteBuffer;
import java.sql.*;
import java.sql.Date;
import java.util.*;
import org.apache.cassandra.db.marshal.CounterColumnType;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlRow;
import org.apache.cassandra.utils.ByteBufferUtil;
public class CResultSet extends AbstractResultSet implements CassandraResultSet
{
private final ColumnDecoder decoder;
private final String keyspace;
private final String columnFamily;
/** The r set iter. */
private Iterator<CqlRow> rSetIter;
int rowNumber = 0;
// the current row key when iterating through results.
private byte[] curRowKey = null;
/** The values. */
private List<TypedColumn> values = new ArrayList<TypedColumn>();
/** The value map. */
private Map<String, TypedColumn> valueMap = new HashMap<String, TypedColumn>();
private final CResultSetMetaData meta;
private boolean wasNull;
/**
* Instantiates a new cassandra result set.
*
* @param resultSet the result set
*/
CResultSet(CqlResult resultSet, ColumnDecoder decoder, String keyspace, String columnFamily)
{
this.decoder = decoder;
this.keyspace = keyspace;
this.columnFamily = columnFamily;
rSetIter = resultSet.getRowsIterator();
meta = new CResultSetMetaData();
}
public byte[] getKey()
{
return curRowKey;
}
public TypedColumn getColumn(int i)
{
return values.get(i);
}
public TypedColumn getColumn(String name)
{
return valueMap.get(name);
}
public void close() throws SQLException
{
valueMap = null;
values = null;
}
private byte[] getBytes(TypedColumn column)
{
ByteBuffer value = (ByteBuffer) column.getValue();
wasNull = value == null;
return value == null ? null : ByteBufferUtil.clone(value).array();
}
public byte[] getBytes(int index) throws SQLException
{
return getBytes(values.get(index - 1));
}
public byte[] getBytes(String name) throws SQLException
{
return getBytes(valueMap.get(name));
}
public Date getDate(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Date getDate(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Date getDate(int arg0, Calendar arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Date getDate(String arg0, Calendar arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public double getDouble(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public double getDouble(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public float getFloat(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public float getFloat(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean getBoolean(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public boolean getBoolean(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
private BigInteger getBigInteger(TypedColumn column)
{
BigInteger value = (BigInteger) column.getValue();
wasNull = value == null;
return value;
}
public BigInteger getBigInteger(int i)
{
return getBigInteger(values.get(i - 1));
}
public BigInteger getBigInteger(String name)
{
return getBigInteger(valueMap.get(name));
}
private int getInt(TypedColumn column) throws SQLException
{
// bit of a hack, this, but asking for getInt seems so common that we should accomodate it
if (column.getValue() instanceof BigInteger)
return getBigInteger(column).intValue();
else if (column.getValue() instanceof Long)
return getLong(column).intValue();
else
throw new SQLException("Non-integer value " + column.getValue());
}
public int getInt(int index) throws SQLException
{
return getInt(values.get(index - 1));
}
public int getInt(String name) throws SQLException
{
return getInt(valueMap.get(name));
}
private Long getLong(TypedColumn column)
{
Long value = (Long) column.getValue();
wasNull = value == null;
return value == null ? 0 : value;
}
public long getLong(int index) throws SQLException
{
return getLong(values.get(index - 1));
}
public long getLong(String name) throws SQLException
{
return getLong(valueMap.get(name));
}
public ResultSetMetaData getMetaData() throws SQLException
{
return meta;
}
private Object getObject(TypedColumn column)
{
Object value = column.getValue();
wasNull = value == null;
return value;
}
public Object getObject(int index) throws SQLException
{
return getObject(values.get(index - 1));
}
public Object getObject(String name) throws SQLException
{
return getObject(valueMap.get(name));
}
public Object getObject(int arg0, Map<String, Class<?>> arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Object getObject(String arg0, Map<String, Class<?>> arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int getRow() throws SQLException
{
return rowNumber;
}
private String getString(TypedColumn column)
{
String value = (String) column.getValue();
wasNull = value == null;
return value == null ? null : value;
}
public String getString(int index) throws SQLException
{
return getString(values.get(index - 1));
}
public String getString(String name) throws SQLException
{
return getString(valueMap.get(name));
}
public Time getTime(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Time getTime(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Time getTime(int arg0, Calendar arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Time getTime(String arg0, Calendar arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Timestamp getTimestamp(int arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Timestamp getTimestamp(String arg0) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Timestamp getTimestamp(int arg0, Calendar arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public Timestamp getTimestamp(String arg0, Calendar arg1) throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public int getType() throws SQLException
{
return ResultSet.TYPE_FORWARD_ONLY;
}
public boolean isAfterLast() throws SQLException
{
return rowNumber == Integer.MAX_VALUE;
}
public boolean isBeforeFirst() throws SQLException
{
return rowNumber == 0;
}
public boolean isClosed() throws SQLException
{
return valueMap == null;
}
public boolean isFirst() throws SQLException
{
return rowNumber == 1;
}
public boolean isLast() throws SQLException
{
return !rSetIter.hasNext();
}
public boolean last() throws SQLException
{
throw new UnsupportedOperationException("method not supported");
}
public <T> T unwrap(Class<T> iface) throws SQLException
{
if (iface.equals(CassandraResultSet.class))
return (T) this;
throw new SQLException("Unsupported unwrap interface: " + iface.getSimpleName());
}
public boolean isWrapperFor(Class<?> iface) throws SQLException
{
return CassandraResultSet.class.isAssignableFrom(iface);
}
public synchronized boolean next() throws SQLException
{
if (!values.isEmpty() || !valueMap.isEmpty())
{
values.clear();
valueMap.clear();
}
if (rSetIter != null && rSetIter.hasNext())
{
CqlRow row = rSetIter.next();
rowNumber++;
curRowKey = row.getKey();
List<Column> cols = row.getColumns();
for (Column col : cols)
{
TypedColumn c = decoder.makeCol(keyspace, columnFamily, col);
values.add(c);
valueMap.put(decoder.colNameAsString(keyspace, columnFamily, col.name), c);
}
return !(values.isEmpty() && valueMap.isEmpty());
}
else
{
rowNumber = Integer.MAX_VALUE;
return false;
}
}
public boolean wasNull() throws SQLException
{
return wasNull;
}
/**
* RSMD implementation. The metadata returned refers to the column
* values, not the column names.
*/
class CResultSetMetaData implements ResultSetMetaData
{
private void checkIndex(int i) throws SQLException
{
if (i >= values.size())
throw new SQLException("Invalid column index " + i);
}
public int getColumnCount() throws SQLException
{
return values.size();
}
public boolean isAutoIncrement(int column) throws SQLException
{
column--;
checkIndex(column);
return values.get(column).getValueType() instanceof CounterColumnType; // todo: check Value is correct.
}
public boolean isCaseSensitive(int column) throws SQLException
{
column--;
checkIndex(column);
TypedColumn tc = values.get(column);
return tc.getValueType().isCaseSensitive();
}
public boolean isSearchable(int column) throws SQLException
{
return false;
}
public boolean isCurrency(int column) throws SQLException
{
column--;
checkIndex(column);
TypedColumn tc = values.get(column);
return tc.getValueType().isCurrency();
}
/** absence is the equivalent of null in Cassandra */
public int isNullable(int column) throws SQLException
{
return ResultSetMetaData.columnNullable;
}
public boolean isSigned(int column) throws SQLException
{
column--;
checkIndex(column);
TypedColumn tc = values.get(column);
return tc.getValueType().isSigned();
}
public int getColumnDisplaySize(int column) throws SQLException
{
column--;
checkIndex(column);
return values.get(column).getValueString().length();
}
public String getColumnLabel(int column) throws SQLException
{
return getColumnName(column);
}
public String getColumnName(int column) throws SQLException
{
column--;
checkIndex(column);
return values.get(column).getNameString();
}
public String getSchemaName(int column) throws SQLException
{
return keyspace;
}
public int getPrecision(int column) throws SQLException
{
column--;
checkIndex(column);
TypedColumn col = values.get(column);
return col.getValueType().getPrecision(col.getValue());
}
public int getScale(int column) throws SQLException
{
column--;
checkIndex(column);
TypedColumn tc = values.get(column);
return tc.getValueType().getScale(tc.getValue());
}
public String getTableName(int column) throws SQLException
{
return columnFamily;
}
public String getCatalogName(int column) throws SQLException
{
throw new SQLFeatureNotSupportedException("Cassandra has no catalogs");
}
public int getColumnType(int column) throws SQLException
{
column--;
checkIndex(column);
return values.get(column).getValueType().getJdbcType();
}
// todo: spec says "database specific type name". this means the abstract type.
public String getColumnTypeName(int column) throws SQLException
{
column--;
checkIndex(column);
return values.get(column).getValueType().getClass().getSimpleName();
}
public boolean isReadOnly(int column) throws SQLException
{
return column == 0;
}
public boolean isWritable(int column) throws SQLException
{
return column > 0;
}
public boolean isDefinitelyWritable(int column) throws SQLException
{
return isWritable(column);
}
public String getColumnClassName(int column) throws SQLException
{
column--;
checkIndex(column);
return values.get(column).getValueType().getType().getName();
}
public <T> T unwrap(Class<T> iface) throws SQLException
{
throw new SQLException("No wrapping implemented");
}
public boolean isWrapperFor(Class<?> iface) throws SQLException
{
return false;
}
}
}

34
drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraPreparedStatement.java
View File

@ -72,21 +72,7 @@ public class CassandraPreparedStatement extends CassandraStatement implements Pr
}
// impl specific methods start here.
// determines which types need to be quoted.
private static boolean needsQuotes(AbstractType type)
{
if (type instanceof ColumnMetaData)
return ((ColumnMetaData)type).needsQuotes();
else
return type == BytesType.instance ||
type == AsciiType.instance ||
type == UTF8Type.instance ||
type == LexicalUUIDType.instance ||
type == TimeUUIDType.instance ||
type == UUIDType.instance;
}
// double quotes strings (in parameters)
private static String makeCqlString(String s)
{
@ -116,7 +102,7 @@ public class CassandraPreparedStatement extends CassandraStatement implements Pr
Object param = params.nextParam();
String stringParam = type == null ? param.toString() : type.toString(param);
stringParam = makeCqlString(stringParam);
if (type == null || needsQuotes(type))
if (type == null || type.needsQuotes())
stringParam = "'" + stringParam + "'";
sb.append(stringParam);
}
@ -153,7 +139,7 @@ public class CassandraPreparedStatement extends CassandraStatement implements Pr
Object param = params.nextParam();
AbstractType type = left ? ltype : rtype;
String stringParam = makeCqlString(type.toString(param));
if (needsQuotes(type))
if (type.needsQuotes())
stringParam = "'" + stringParam + "'";
sb.append(stringParam);
}
@ -194,12 +180,12 @@ public class CassandraPreparedStatement extends CassandraStatement implements Pr
String columnFamily = connection.getColumnFamily(cql);
ParameterIterator params = new ParameterIterator();
String left = cql.substring(0, pivot);
AbstractType leftType = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Comparator, null);
AbstractType leftType = connection.decoder.getComparator(keyspace, columnFamily);
if (leftType == null)
throw new SQLException("Could not find comparator for " + keyspace + "." + columnFamily);
left = applySimpleBindings(left, leftType, params);
String right = cql.substring(pivot);
AbstractType keyVald = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Key, null);
AbstractType keyVald = connection.decoder.getKeyValidator(keyspace, columnFamily);
if (keyVald == null)
throw new SQLException("Could not find key validator for " + keyspace + "." + columnFamily);
right = applySimpleBindings(right, keyVald, params);
@ -213,12 +199,12 @@ public class CassandraPreparedStatement extends CassandraStatement implements Pr
String columnFamily = connection.getColumnFamily(cql);
ParameterIterator params = new ParameterIterator();
String left = cql.substring(0, pivot);
AbstractType leftType = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Comparator, null);
AbstractType leftType = connection.decoder.getComparator(keyspace, columnFamily);
if (leftType == null)
throw new SQLException("Could not find comparator for " + keyspace + "." + columnFamily);
left = applySimpleBindings(left, leftType, params);
String right = cql.substring(pivot);
AbstractType keyVald = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Key, null);
AbstractType keyVald = connection.decoder.getKeyValidator(keyspace, columnFamily);
if (keyVald == null)
throw new SQLException("Could not find key validator for " + keyspace + "." + columnFamily);
right = applySimpleBindings(right, keyVald, params);
@ -234,15 +220,15 @@ public class CassandraPreparedStatement extends CassandraStatement implements Pr
String columnFamily = connection.getColumnFamily(cql);
ParameterIterator params = new ParameterIterator();
String left = cql.substring(0, pivot);
AbstractType leftComp = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Comparator, null);
AbstractType leftComp = connection.decoder.getComparator(keyspace, columnFamily);
if (leftComp == null)
throw new SQLException("Could not find comparator for " + keyspace + "." + columnFamily);
AbstractType leftVald = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Validator, null);
AbstractType leftVald = connection.decoder.getComparator(keyspace, columnFamily);
if (leftVald == null)
throw new SQLException("Could not find validator for " + keyspace + "." + columnFamily);
left = applyDualBindings(left, leftComp, leftVald, params);
String right = cql.substring(pivot);
AbstractType keyVald = connection.decoder.getComparator(keyspace, columnFamily, ColumnDecoder.Specifier.Key, null);
AbstractType keyVald = connection.decoder.getKeyValidator(keyspace, columnFamily);
if (keyVald == null)
throw new SQLException("Could not find key validator for " + keyspace + "." + columnFamily);
right = applySimpleBindings(right, keyVald, params);

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

File diff suppressed because it is too large Load Diff

56
drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraResultSetMetaData.java
View File

@ -1,56 +0,0 @@
package org.apache.cassandra.cql.jdbc;
/*
*
* 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.
*
*/
import java.sql.SQLException;
/**
* ResultSetMetaData give lots of nice detailed type inforamtion about column values.
* This interface aims to do the same thing but distinguishes column names and values.
*/
public interface CassandraResultSetMetaData
{
/** exposes row key */
public byte[] getKey();
// the rest of these methods have similar calls in java.sql.ResultSetMetaData.
public boolean isNameCaseSensitive(int column) throws SQLException;
public boolean isNameCurrency(int column) throws SQLException;
public boolean isNameSigned(int column) throws SQLException;
public int getNameDisplaySize(int column) throws SQLException;
public int getNamePrecision(int column) throws SQLException;
public int getNameScale(int column) throws SQLException;
public int getNameType(int column) throws SQLException;
public String getNameTypeName(int column) throws SQLException;
public String getNameClassName(int column) throws SQLException;
public boolean isValueCaseSensitive(int column) throws SQLException;
public boolean isValueCurrency(int column) throws SQLException;
public boolean isValueSigned(int column) throws SQLException;
public int getValueDisplaySize(int column) throws SQLException;
public int getValuePrecision(int column) throws SQLException;
public int getValueScale(int column) throws SQLException;
public int getValueType(int column) throws SQLException;
public String getValueTypeName(int column) throws SQLException;
public String getValueClassName(int column) throws SQLException;
}

2
drivers/java/src/org/apache/cassandra/cql/jdbc/CassandraStatement.java
View File

@ -235,7 +235,7 @@ class CassandraStatement implements Statement
{
CqlResult rSet = connection.execute(query);
// todo: encapsulate.
return new CassandraResultSet(rSet, connection.decoder, connection.curKeyspace, connection.curColumnFamily);
return new CResultSet(rSet, connection.decoder, connection.curKeyspace, connection.curColumnFamily);
}
catch (InvalidRequestException e)
{

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

@ -21,13 +21,14 @@ package org.apache.cassandra.cql.jdbc;
*/
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.ColumnDef;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -36,160 +37,76 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
/** Decodes columns from bytes into instances of their respective expected types. */
class ColumnDecoder
/**
* Decodes columns from bytes into instances of their respective expected types.
*/
class ColumnDecoder
{
private static final Logger logger = LoggerFactory.getLogger(ColumnDecoder.class);
private static final String MapFormatString = "%s.%s.%s.%s";
// basically denotes column or value.
enum Specifier
{
Comparator,
Validator,
Key,
ColumnSpecific
}
private Map<String, CfDef> cfDefs = new HashMap<String, CfDef>();
// cache the comparators for efficiency.
private Map<String, AbstractType> comparators = new HashMap<String, AbstractType>();
/** is specific per set of keyspace definitions. */
private final Map<String, CFMetaData> metadata = new HashMap<String, CFMetaData>();
/**
* is specific per set of keyspace definitions.
*/
public ColumnDecoder(List<KsDef> defs)
{
for (KsDef ks : defs)
for (KsDef ks : defs)
{
for (CfDef cf : ks.getCf_defs())
{
cfDefs.put(String.format("%s.%s", ks.getName(), cf.getName()), cf);
for (ColumnDef cd : cf.getColumn_metadata())
try
{
try
{
// prefill the validators (because they aren't kept in a convenient lookup map and we don't
// want to iterate over the list for every miss in getComparator.
comparators.put(String.format(MapFormatString,
ks.getName(),
cf.getName(),
Specifier.ColumnSpecific.name(),
ByteBufferUtil.bytesToHex(cd.bufferForName())),
FBUtilities.getComparator(cd.getValidation_class()));
}
catch (ConfigurationException ex) {
throw new RuntimeException(ex);
}
metadata.put(String.format("%s.%s", ks.getName(), cf.getName()), CFMetaData.convertToCFMetaData(cf));
}
catch (InvalidRequestException e)
{
throw new RuntimeException(e);
}
catch (ConfigurationException e)
{
throw new RuntimeException(e);
}
}
}
}
/**
* @param keyspace ALWAYS specify
* @param columnFamily ALWAYS specify
* @param specifier ALWAYS specify
* @param def avoids additional map lookup if specified. null is ok though.
* @return
*/
AbstractType getComparator(String keyspace, String columnFamily, Specifier specifier, CfDef def)
AbstractType getComparator(String keyspace, String columnFamily)
{
return getComparator(keyspace, columnFamily, null, specifier, def);
}
// same as above, but can get column-specific validators.
AbstractType getComparator(String keyspace, String columnFamily, byte[] column, Specifier specifier, CfDef def)
{
// check cache first.
String key = String.format(MapFormatString,
keyspace,
columnFamily,
specifier.name(),
FBUtilities.bytesToHex(column == null ? new byte[] {} : column));
AbstractType comparator = comparators.get(key);
// make and put in cache.
if (comparator == null)
{
if (def == null)
def = cfDefs.get(String.format("%s.%s", keyspace, columnFamily));
if (def == null)
// no point in proceeding. these values are bad.
return null;
try
{
switch (specifier)
{
case Key:
comparator = FBUtilities.getComparator(def.getKey_validation_class());
break;
case ColumnSpecific:
// if we get here this means there is no column-specific validator, so fall through to the default.
case Validator:
comparator = FBUtilities.getComparator(def.getDefault_validation_class());
break;
case Comparator:
default:
comparator = FBUtilities.getComparator(def.getComparator_type());
break;
}
comparators.put(key, comparator);
}
catch (ConfigurationException ex)
{
throw new RuntimeException(ex);
}
}
return comparator;
return metadata.get(String.format("%s.%s", keyspace, columnFamily)).comparator;
}
/**
* uses the AbstractType to map a column name to a string. Relies on AT.fromString() and AT.getString()
* @param keyspace
* @param columnFamily
* @param name
* @return
*/
public String colNameAsString(String keyspace, String columnFamily, String name)
AbstractType getNameType(String keyspace, String columnFamily, ByteBuffer name)
{
AbstractType comparator = getComparator(keyspace, columnFamily, Specifier.Comparator, null);
ByteBuffer bb = comparator.fromString(name);
return comparator.getString(bb);
CFMetaData md = metadata.get(String.format("%s.%s", keyspace, columnFamily));
return md.comparator;
}
/**
* uses the AbstractType to map a column name to a string.
* @param keyspace
* @param columnFamily
* @param name
* @return
*/
public String colNameAsString(String keyspace, String columnFamily, byte[] name)
AbstractType getValueType(String keyspace, String columnFamily, ByteBuffer name)
{
AbstractType comparator = getComparator(keyspace, columnFamily, Specifier.Comparator, null);
return comparator.getString(ByteBuffer.wrap(name));
CFMetaData md = metadata.get(String.format("%s.%s", keyspace, columnFamily));
ColumnDefinition cd = md.getColumnDefinition(name);
return cd == null ? md.getDefaultValidator() : cd.getValidator();
}
/**
* converts a column value to a string.
* @param value
* @return
*/
public static String colValueAsString(Object value) {
if (value instanceof String)
return (String)value;
else if (value instanceof byte[])
return ByteBufferUtil.bytesToHex(ByteBuffer.wrap((byte[])value));
else
return value == null ? null : value.toString();
public AbstractType getKeyValidator(String keyspace, String columnFamily)
{
return metadata.get(String.format("%s.%s", keyspace, columnFamily)).getKeyValidator();
}
/** uses the AbstractType to map a column name to a string. */
public String colNameAsString(String keyspace, String columnFamily, ByteBuffer name)
{
AbstractType comparator = getNameType(keyspace, columnFamily, name);
return comparator.getString(name);
}
/** constructs a typed column */
public TypedColumn makeCol(String keyspace, String columnFamily, byte[] name, byte[] value)
public TypedColumn makeCol(String keyspace, String columnFamily, Column column)
{
CfDef cfDef = cfDefs.get(String.format("%s.%s", keyspace, columnFamily));
AbstractType comparator = getComparator(keyspace, columnFamily, Specifier.Comparator, cfDef);
AbstractType validator = getComparator(keyspace, columnFamily, name, Specifier.ColumnSpecific, null);
return new TypedColumn(comparator, name, validator, value);
return new TypedColumn(column,
getNameType(keyspace, columnFamily, column.name),
getValueType(keyspace, columnFamily, column.name));
}
}

38
drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnMetaData.java
View File

@ -1,38 +0,0 @@
package org.apache.cassandra.cql.jdbc;
/*
*
* 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.
*
*/
/**
* This interface lets AbstractTypes derived outside of the cassandra project answer JDBC questions
* relating to type. Only bother implmenting this on your AbstractTypes if you're going to use JDBC
* and ResultSetMetaData.
*/
public interface ColumnMetaData
{
public boolean isSigned();
public boolean isCaseSensitive();
public boolean isCurrency();
public int getPrecision();
public int getScale();
public int getType();
public boolean needsQuotes();
}

56
drivers/java/src/org/apache/cassandra/cql/jdbc/TypedColumn.java
View File

@ -22,37 +22,36 @@ package org.apache.cassandra.cql.jdbc;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.thrift.Column;
import java.nio.ByteBuffer;
class TypedColumn<N, V>
class TypedColumn
{
private final N name;
private final V value;
// we cache the string versions of the byte buffers here. It turns out that {N|V}.toString() isn't always the same
// (a good example is byte buffers) as the stringified versions supplied by the AbstractTypes.
private final Column rawColumn;
// we cache the frequently-accessed forms: java object for value, String for name.
// Note that {N|V}.toString() isn't always the same as Type.getString
// (a good example is byte buffers).
private final Object value;
private final String nameString;
private final String valueString;
private final AbstractType<V> validator;
public TypedColumn(AbstractType<N> comparator, byte[] name, AbstractType<V> validator, byte[] value)
private final AbstractType nameType, valueType;
public TypedColumn(Column column, AbstractType comparator, AbstractType validator)
{
ByteBuffer bbName = ByteBuffer.wrap(name);
ByteBuffer bbValue = value == null ? null : ByteBuffer.wrap(value);
this.name = comparator.compose(bbName);
this.value = value == null ? null : validator.compose(bbValue);
nameString = comparator.getString(bbName);
valueString = value == null ? null : validator.getString(bbValue);
this.validator = validator;
rawColumn = column;
this.value = column.value == null ? null : validator.compose(column.value);
nameString = comparator.getString(column.name);
nameType = comparator;
valueType = validator;
}
public Column getRawColumn()
{
return rawColumn;
}
public N getName()
{
return name;
}
public V getValue()
public Object getValue()
{
return value;
}
@ -64,11 +63,16 @@ class TypedColumn<N, V>
public String getValueString()
{
return valueString;
return valueType.getString(rawColumn.value);
}
public AbstractType<V> getValidator()
public AbstractType getNameType()
{
return validator;
return nameType;
}
public AbstractType getValueType()
{
return valueType;
}
}

36
drivers/java/src/org/apache/cassandra/cql/jdbc/Utils.java
View File

@ -63,40 +63,4 @@ class Utils
return ByteBuffer.wrap(byteArray.toByteArray());
}
static int getJdbcType(AbstractType type) throws SQLException
{
if (type instanceof ColumnMetaData)
return ((ColumnMetaData)type).getType();
else if (type == IntegerType.instance)
return Types.BIGINT;
else if (type.getType().equals(Long.class))
return Types.BIGINT; // not the best fit.
else if (type.getType().equals(String.class))
return Types.VARCHAR;
else if (type.getType().equals(UUID.class))
return Types.TIMESTAMP;
else if (type == BytesType.instance)
return Types.BINARY;
else
throw new SQLException("Uninterpretable JDBC type " + type.getClass().getName());
}
static boolean isTypeSigned(AbstractType type)
{
if (type == IntegerType.instance || type == LongType.instance)
return true;
else if (type instanceof ColumnMetaData)
return ((ColumnMetaData)type).isSigned();
else
return false;
}
static int getTypeScale(AbstractType type)
{
if (type instanceof ColumnMetaData)
return ((ColumnMetaData)type).getScale();
else
return 0;
}
}

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

@ -25,27 +25,18 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.sql.*;
import java.util.Arrays;
import org.apache.cassandra.cql.jdbc.CassandraResultSetMetaData;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.utils.FBUtilities;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.utils.FBUtilities;
import static junit.framework.Assert.assertEquals;
/**
* Test case for unit test of various methods of JDBC implementation.
*/
@ -71,9 +62,9 @@ public class JdbcDriverTest extends EmbeddedServiceBase
"UPDATE JdbcInteger SET 1 = 11, 2 = 22, 42='fortytwo' WHERE KEY = '" + jsmith + "'",
"UPDATE JdbcInteger SET 3 = 33, 4 = 44 WHERE KEY = '" + jsmith + "'",
"UPDATE JdbcLong SET 1 = 11, 2 = 22 WHERE KEY = '" + jsmith + "'",
"UPDATE JdbcAscii SET 'first' = 'firstrec', 'last' = 'lastrec' WHERE key = '" + jsmith + "'",
"UPDATE JdbcAscii SET 'first' = 'firstrec', last = 'lastrec' WHERE key = '" + jsmith + "'",
String.format("UPDATE JdbcBytes SET '%s' = '%s', '%s' = '%s' WHERE key = '%s'", first, firstrec, last, lastrec, jsmith),
"UPDATE JdbcUtf8 SET 'first' = 'firstrec', 'fortytwo' = '42', 'last' = 'lastrec' WHERE key = '" + jsmith + "'",
"UPDATE JdbcUtf8 SET 'first' = 'firstrec', fortytwo = '42', last = 'lastrec' WHERE key = '" + jsmith + "'",
};
for (String q : inserts)
{
@ -91,32 +82,25 @@ public class JdbcDriverTest extends EmbeddedServiceBase
private static void expectedMetaData(ResultSetMetaData md, int col, String colClass, String table, String schema,
String label, int type, String typeName, boolean signed, boolean caseSensitive) throws SQLException
{
assert colClass.equals(md.getColumnClassName(col)); // full class name of type<T>
assert table.equals(md.getTableName(col));
assert schema.equals(md.getSchemaName(col));
assert label.equals(md.getColumnLabel(col)) : "expected " + label + " got " + md.getColumnLabel(col);
assert label.equals(md.getColumnName(col));
assert type == md.getColumnType(col);
assert typeName.equals(md.getColumnTypeName(col)) : "expected " + typeName + " got " + md.getColumnTypeName(col); // simple name of abstract type.
assert md.isSigned(col) == signed;
assert md.isCaseSensitive(col) == caseSensitive;
assertEquals(colClass, md.getColumnClassName(col)); // full class name of type<T>
assertEquals(table, md.getTableName(col));
assertEquals(schema, md.getSchemaName(col));
assertEquals(label, md.getColumnLabel(col));
assertEquals(label, md.getColumnName(col));
assertEquals(type, md.getColumnType(col));
assertEquals(typeName, md.getColumnTypeName(col));
assertEquals(signed, md.isSigned(col));
assertEquals(caseSensitive, md.isCaseSensitive(col));
}
private static void expectedMetaData(CassandraResultSetMetaData md, int col,
String nameClass, int nameType, String nameTypeName, boolean nameSigned, boolean nameCaseSense,
private static void expectedMetaData(ResultSetMetaData md, int col,
String valuClass, int valuType, String valuTypeName, boolean valuSigned, boolean valuCaseSense) throws SQLException
{
assert nameClass.equals(md.getNameClassName(col));
assert nameType == md.getNameType(col);
assert nameTypeName.equals(md.getNameTypeName(col));
assert nameSigned == md.isNameSigned(col);
assert nameCaseSense == md.isNameCaseSensitive(col);
assert valuClass.equals(md.getValueClassName(col));
assert valuType == md.getValueType(col);
assert valuTypeName.equals(md.getValueTypeName(col));
assert valuSigned == md.isValueSigned(col);
assert valuCaseSense == md.isValueCaseSensitive(col);
assertEquals(valuClass, md.getColumnClassName(col));
assertEquals(valuType, md.getColumnType(col));
assertEquals(valuTypeName, md.getColumnTypeName(col));
assertEquals(valuSigned, md.isSigned(col));
assertEquals(valuCaseSense, md.isCaseSensitive(col));
}
@Test
@ -124,11 +108,10 @@ public class JdbcDriverTest extends EmbeddedServiceBase
{
String key = FBUtilities.bytesToHex("Integer".getBytes());
Statement stmt = con.createStatement();
stmt.executeUpdate("update JdbcInteger set 1=1111, 2=2222, 42='fortytwofortytwo' where key='" + key + "'");
stmt.executeUpdate("update JdbcInteger set 1=36893488147419103232, 42='fortytwofortytwo' where key='" + key + "'");
ResultSet rs = stmt.executeQuery("select 1, 2, 42 from JdbcInteger where key='" + key + "'");
assert rs.next();
assert rs.getInt("1") == 1111;
assert rs.getInt("2") == 2222;
assert rs.getObject("1").equals(new BigInteger("36893488147419103232"));
assert rs.getString("42").equals("fortytwofortytwo") : rs.getString("42");
ResultSetMetaData md = rs.getMetaData();
@ -137,8 +120,8 @@ public class JdbcDriverTest extends EmbeddedServiceBase
expectedMetaData(md, 2, BigInteger.class.getName(), "JdbcInteger", "Keyspace1", "2", Types.BIGINT, IntegerType.class.getSimpleName(), true, false);
expectedMetaData(md, 3, String.class.getName(), "JdbcInteger", "Keyspace1", "42", Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true);
stmt.executeUpdate("update JdbcUtf8 set 'a'='aa', 'b'='bb', 'fortytwo'='4242' where key='" + key + "'");
rs = stmt.executeQuery("select 'a', 'b', 'fortytwo' from JdbcUtf8 where key='" + key + "'");
stmt.executeUpdate("update JdbcUtf8 set a='aa', b='bb', fortytwo='4242' where key='" + key + "'");
rs = stmt.executeQuery("select a, b, fortytwo from JdbcUtf8 where key='" + key + "'");
assert rs.next();
assert rs.getString("a").equals("aa");
assert rs.getString("b").equals("bb");
@ -149,32 +132,7 @@ public class JdbcDriverTest extends EmbeddedServiceBase
expectedMetaData(md, 2, String.class.getName(), "JdbcUtf8", "Keyspace1", "b", Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true);
expectedMetaData(md, 3, BigInteger.class.getName(), "JdbcUtf8", "Keyspace1", "fortytwo", Types.BIGINT, IntegerType.class.getSimpleName(), true, false);
}
@Test
public void testIntegerMetadata() throws SQLException
{
String key = FBUtilities.bytesToHex("Integer".getBytes());
Statement stmt = con.createStatement();
stmt.executeUpdate("UPDATE JdbcInteger SET 1=111, 2=222 WHERE KEY = '" + key + "'");
ResultSet rs = stmt.executeQuery("SELECT 1, 2 from JdbcInteger WHERE KEY = '" + key + "'");
assert rs.next();
assert rs.getInt("1") == 111;
assert rs.getInt("2") == 222;
ResultSetMetaData md = rs.getMetaData();
assert md.getColumnCount() == 2;
expectedMetaData(md, 1, BigInteger.class.getName(), "JdbcInteger", "Keyspace1", "1", Types.BIGINT, IntegerType.class.getSimpleName(), true, false);
expectedMetaData(md, 2, BigInteger.class.getName(), "JdbcInteger", "Keyspace1", "2", Types.BIGINT, IntegerType.class.getSimpleName(), true, false);
CassandraResultSetMetaData cmd = md.unwrap(CassandraResultSetMetaData.class);
for (int i = 0; i < md.getColumnCount(); i++)
expectedMetaData(
cmd, i+1,
BigInteger.class.getName(), Types.BIGINT, IntegerType.class.getSimpleName(), true, false,
BigInteger.class.getName(), Types.BIGINT, IntegerType.class.getSimpleName(), true, false);
}
@Test
public void testLongMetadata() throws SQLException
{
@ -188,27 +146,23 @@ public class JdbcDriverTest extends EmbeddedServiceBase
ResultSetMetaData md = rs.getMetaData();
assert md.getColumnCount() == 2;
expectedMetaData(md, 1, Long.class.getName(), "JdbcLong", "Keyspace1", "1", Types.BIGINT, LongType.class.getSimpleName(), true, false);
expectedMetaData(md, 2, Long.class.getName(), "JdbcLong", "Keyspace1", "2", Types.BIGINT, LongType.class.getSimpleName(), true, false);
expectedMetaData(md, 1, Long.class.getName(), "JdbcLong", "Keyspace1", "1", Types.INTEGER, LongType.class.getSimpleName(), true, false);
expectedMetaData(md, 2, Long.class.getName(), "JdbcLong", "Keyspace1", "2", Types.INTEGER, LongType.class.getSimpleName(), true, false);
CassandraResultSetMetaData cmd = md.unwrap(CassandraResultSetMetaData.class);
for (int i = 0; i < md.getColumnCount(); i++)
expectedMetaData(
cmd, i+1,
Long.class.getName(), Types.BIGINT, LongType.class.getSimpleName(), true, false,
Long.class.getName(), Types.BIGINT, LongType.class.getSimpleName(), true, false);
expectedMetaData(md, i + 1, Long.class.getName(), Types.INTEGER, LongType.class.getSimpleName(), true, false);
}
@Test
public void testStringMetadata() throws SQLException
{
String aKey = FBUtilities.bytesToHex("ascii".getBytes());
String uKey = FBUtilities.bytesToHex("utf8".getBytes());
Statement stmt = con.createStatement();
stmt.executeUpdate("UPDATE JdbcAscii SET 'a'='aa', 'b'='bb' WHERE KEY = '" + aKey + "'");
stmt.executeUpdate("UPDATE JdbcUtf8 SET 'a'='aa', 'b'='bb' WHERE KEY = '" + uKey + "'");
ResultSet rs0 = stmt.executeQuery("SELECT 'a', 'b' FROM JdbcAscii WHERE KEY = '" + aKey + "'");
ResultSet rs1 = stmt.executeQuery("SELECT 'a', 'b' FROM JdbcUtf8 WHERE KEY = '" + uKey + "'");
stmt.executeUpdate("UPDATE JdbcAscii SET a='aa', b='bb' WHERE KEY = '" + aKey + "'");
stmt.executeUpdate("UPDATE JdbcUtf8 SET a='aa', b='bb' WHERE KEY = '" + uKey + "'");
ResultSet rs0 = stmt.executeQuery("SELECT a, b FROM JdbcAscii WHERE KEY = '" + aKey + "'");
ResultSet rs1 = stmt.executeQuery("SELECT a, b FROM JdbcUtf8 WHERE KEY = '" + uKey + "'");
for (ResultSet rs : new ResultSet[] { rs0, rs1 })
{
assert rs.next();
@ -224,18 +178,24 @@ public class JdbcDriverTest extends EmbeddedServiceBase
assert md.getColumnCount() == 2;
expectedMetaData(md, 1, String.class.getName(), "JdbcUtf8", "Keyspace1", "a", Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true);
expectedMetaData(md, 2, String.class.getName(), "JdbcUtf8", "Keyspace1", "b", Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true);
CassandraResultSetMetaData cmd0 = rs0.getMetaData().unwrap(CassandraResultSetMetaData.class);
CassandraResultSetMetaData cmd1 = rs1.getMetaData().unwrap(CassandraResultSetMetaData.class);
for (int i = 0; i < 2; i++)
{
expectedMetaData(cmd0, i+1,
String.class.getName(), Types.VARCHAR, AsciiType.class.getSimpleName(), false, true,
String.class.getName(), Types.VARCHAR, AsciiType.class.getSimpleName(), false, true);
expectedMetaData(cmd1, i+1,
String.class.getName(), Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true,
String.class.getName(), Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true);
expectedMetaData(rs0.getMetaData(),
i + 1,
String.class.getName(),
Types.VARCHAR,
AsciiType.class.getSimpleName(),
false,
true);
expectedMetaData(rs1.getMetaData(),
i + 1,
String.class.getName(),
Types.VARCHAR,
UTF8Type.class.getSimpleName(),
false,
true);
}
}
@ -268,13 +228,10 @@ public class JdbcDriverTest extends EmbeddedServiceBase
expectedMetaData(md, 1, ByteBuffer.class.getName(), "JdbcBytes", "Keyspace1", FBUtilities.bytesToHex(a), Types.BINARY, BytesType.class.getSimpleName(), false, false);
expectedMetaData(md, 2, ByteBuffer.class.getName(), "JdbcBytes", "Keyspace1", FBUtilities.bytesToHex(b), Types.BINARY, BytesType.class.getSimpleName(), false, false);
CassandraResultSetMetaData cmd = md.unwrap(CassandraResultSetMetaData.class);
for (int i = 0; i < md.getColumnCount(); i++)
expectedMetaData(cmd, i+1,
ByteBuffer.class.getName(), Types.BINARY, BytesType.class.getSimpleName(), false, false,
ByteBuffer.class.getName(), Types.BINARY, BytesType.class.getSimpleName(), false, false);
expectedMetaData(md, i + 1, ByteBuffer.class.getName(), Types.BINARY, BytesType.class.getSimpleName(), false, false);
}
@Test
public void testWithStatementBytesType() throws SQLException
{
@ -305,13 +262,13 @@ public class JdbcDriverTest extends EmbeddedServiceBase
selectQ = "SELECT 1, 2 FROM JdbcLong WHERE KEY='" + jsmith + "'";
checkResultSet(stmt.executeQuery(selectQ), "Long", 1, "1", "2");
selectQ = "SELECT 'first', 'last' FROM JdbcAscii WHERE KEY='" + jsmith + "'";
selectQ = "SELECT 'first', last FROM JdbcAscii WHERE KEY='" + jsmith + "'";
checkResultSet(stmt.executeQuery(selectQ), "String", 1, "first", "last");
selectQ = String.format("SELECT '%s', '%s' FROM JdbcBytes WHERE KEY='%s'", first, last, jsmith);
checkResultSet(stmt.executeQuery(selectQ), "Bytes", 1, first, last);
selectQ = "SELECT 'first', 'last' FROM JdbcUtf8 WHERE KEY='" + jsmith + "'";
selectQ = "SELECT 'first', last FROM JdbcUtf8 WHERE KEY='" + jsmith + "'";
checkResultSet(stmt.executeQuery(selectQ), "String", 1, "first", "last");
}
@ -344,13 +301,13 @@ public class JdbcDriverTest extends EmbeddedServiceBase
selectQ = "SELECT 1, 2 FROM JdbcLong WHERE KEY='" + jsmith + "'";
checkResultSet(executePreparedStatementWithResults(con, selectQ), "Long", 1, "1", "2");
selectQ = "SELECT 'first', 'last' FROM JdbcAscii WHERE KEY='" + jsmith + "'";
selectQ = "SELECT 'first', last FROM JdbcAscii WHERE KEY='" + jsmith + "'";
checkResultSet(executePreparedStatementWithResults(con, selectQ), "String", 1, "first", "last");
selectQ = String.format("SELECT '%s', '%s' FROM JdbcBytes WHERE KEY='%s'", first, last, jsmith);
checkResultSet(executePreparedStatementWithResults(con, selectQ), "Bytes", 1, first, last);
selectQ = "SELECT 'first', 'last' FROM JdbcUtf8 WHERE KEY='" + jsmith + "'";
selectQ = "SELECT 'first', last FROM JdbcUtf8 WHERE KEY='" + jsmith + "'";
checkResultSet(executePreparedStatementWithResults(con, selectQ), "String", 1, "first", "last");
}
@ -383,7 +340,7 @@ public class JdbcDriverTest extends EmbeddedServiceBase
"DELETE 'first' FROM JdbcAscii WHERE KEY='" + jsmith + "'",
"SELECT 'first' FROM JdbcAscii WHERE KEY='" + jsmith + "'",
"SELECT 'last' FROM JdbcAscii WHERE KEY='" + jsmith + "'",
"SELECT last FROM JdbcAscii WHERE KEY='" + jsmith + "'",
String.format("DELETE '%s' FROM JdbcBytes WHERE KEY='%s'", first, jsmith),
String.format("SELECT '%s' FROM JdbcBytes WHERE KEY='%s'", first, jsmith),
@ -391,7 +348,7 @@ public class JdbcDriverTest extends EmbeddedServiceBase
"DELETE 'first' FROM JdbcUtf8 WHERE KEY='" + jsmith + "'",
"SELECT 'first' FROM JdbcUtf8 WHERE KEY='" + jsmith + "'",
"SELECT 'last' FROM JdbcUtf8 WHERE KEY='" + jsmith + "'",
"SELECT last FROM JdbcUtf8 WHERE KEY='" + jsmith + "'",
};
for (int i = 0; i < statements.length/3; i++)
@ -455,9 +412,7 @@ public class JdbcDriverTest extends EmbeddedServiceBase
actualRows++;
for (int c = 0; c < cols.length; c++)
{
// getString and getObject should always work.
assert rs.getString(cols[c]) != null;
assert rs.getString(c+1) != null;
// getObject should always work.
assert rs.getObject(cols[c]) != null;
assert rs.getObject(c+1) != null;

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

@ -984,6 +984,11 @@ public final class CFMetaData
comparator.validate(cf_def.key_alias);
}
public ColumnDefinition getColumnDefinition(ByteBuffer name)
{
return column_metadata.get(name);
}
@Override
public String toString()
{

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

@ -80,6 +80,7 @@ public class QueryProcessor
AbstractType<?> comparator = select.getComparator(keyspace);
List<ReadCommand> commands = new ArrayList<ReadCommand>();
assert select.getKeys().size() == 1;
CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false);
// ...of a list of column names
@ -138,11 +139,20 @@ public class QueryProcessor
private static List<org.apache.cassandra.db.Row> multiRangeSlice(String keyspace, SelectStatement select)
throws TimedOutException, UnavailableException, InvalidRequestException
{
List<org.apache.cassandra.db.Row> rows;
IPartitioner<?> p = StorageService.getPartitioner();
AbstractType<?> keyType = DatabaseDescriptor.getCFMetaData(keyspace,
select.getColumnFamily()).getKeyValidator();
ByteBuffer startKey = (select.getKeyStart() != null) ? select.getKeyStart().getByteBuffer(keyType) : (new Term()).getByteBuffer();
ByteBuffer finishKey = (select.getKeyFinish() != null) ? select.getKeyFinish().getByteBuffer(keyType) : (new Term()).getByteBuffer();
IPartitioner<?> p = StorageService.getPartitioner();
ByteBuffer startKey = (select.getKeyStart() != null)
? select.getKeyStart().getByteBuffer(keyType)
: (new Term()).getByteBuffer();
ByteBuffer finishKey = (select.getKeyFinish() != null)
? select.getKeyFinish().getByteBuffer(keyType)
: (new Term()).getByteBuffer();
AbstractBounds bounds = new Bounds(p.getToken(startKey), p.getToken(finishKey));
CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false);
@ -151,14 +161,18 @@ public class QueryProcessor
SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, comparator);
validateSlicePredicate(metadata, thriftSlicePredicate);
int limit = select.isKeyRange() && select.getKeyStart() != null
? select.getNumRecords() + 1
: select.getNumRecords();
try
{
return StorageProxy.getRangeSlice(new RangeSliceCommand(keyspace,
rows = StorageProxy.getRangeSlice(new RangeSliceCommand(keyspace,
select.getColumnFamily(),
null,
thriftSlicePredicate,
bounds,
select.getNumRecords()),
limit),
select.getConsistencyLevel());
}
catch (IOException e)
@ -173,6 +187,23 @@ public class QueryProcessor
{
throw new TimedOutException();
}
// if start key was set and relation was "greater than"
if (select.getKeyStart() != null && !select.includeStartKey())
{
if (rows.get(0).key.key.equals(startKey))
rows.remove(0);
}
// if finish key was set and relation was "less than"
if (select.getKeyFinish() != null && !select.includeFinishKey())
{
int lastIndex = rows.size() - 1;
if (rows.get(lastIndex).key.key.equals(finishKey))
rows.remove(lastIndex);
}
return rows.subList(0, select.getNumRecords() < rows.size() ? select.getNumRecords() : rows.size());
}
private static List<org.apache.cassandra.db.Row> getIndexedSlices(String keyspace, SelectStatement select)

12
src/java/org/apache/cassandra/cql/SelectStatement.java
View File

@ -126,7 +126,17 @@ public class SelectStatement
{
return isCountOper;
}
public boolean includeStartKey()
{
return clause.includeStartKey();
}
public boolean includeFinishKey()
{
return clause.includeFinishKey();
}
public AbstractType getComparator(String keyspace)
{
return DatabaseDescriptor.getComparator(keyspace, columnFamily);

19
src/java/org/apache/cassandra/cql/WhereClause.java
View File

@ -33,7 +33,8 @@ public class WhereClause
private List<Term> keys = new ArrayList<Term>();
private Term startKey, finishKey;
private List<Relation> columns = new ArrayList<Relation>();
private boolean includeStartKey = false, includeFinishKey = false;
/**
* Create a new WhereClause with the first parsed relation.
*
@ -61,9 +62,15 @@ public class WhereClause
if (relation.operator().equals(RelationType.EQ))
keys.add(relation.getValue());
else if ((relation.operator().equals(RelationType.GT) || relation.operator().equals(RelationType.GTE)))
{
startKey = relation.getValue();
includeStartKey = relation.operator().equals(RelationType.GTE);
}
else if ((relation.operator().equals(RelationType.LT) || relation.operator().equals(RelationType.LTE)))
{
finishKey = relation.getValue();
includeFinishKey = relation.operator().equals(RelationType.LTE);
}
}
else
@ -108,4 +115,14 @@ public class WhereClause
{
return keys;
}
public boolean includeStartKey()
{
return includeStartKey;
}
public boolean includeFinishKey()
{
return includeFinishKey;
}
}

11
src/java/org/apache/cassandra/db/BinaryMemtable.java
View File

@ -35,6 +35,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
@ -83,7 +84,7 @@ public class BinaryMemtable implements IFlushable
if (!isFrozen)
{
isFrozen = true;
cfs.submitFlush(this, new CountDownLatch(1));
cfs.submitFlush(this, new CountDownLatch(1), null);
cfs.switchBinaryMemtable(key, buffer);
}
else
@ -122,10 +123,10 @@ public class BinaryMemtable implements IFlushable
return keys;
}
private SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys) throws IOException
private SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys, ReplayPosition context) throws IOException
{
logger.info("Writing " + this);
SSTableWriter writer = cfs.createFlushWriter(sortedKeys.size(), DatabaseDescriptor.getBMTThreshold());
SSTableWriter writer = cfs.createFlushWriter(sortedKeys.size(), DatabaseDescriptor.getBMTThreshold(), context);
for (DecoratedKey key : sortedKeys)
{
@ -138,7 +139,7 @@ public class BinaryMemtable implements IFlushable
return sstable;
}
public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer)
public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context)
{
sorter.execute(new Runnable()
{
@ -149,7 +150,7 @@ public class BinaryMemtable implements IFlushable
{
public void runMayThrow() throws IOException
{
cfs.addSSTable(writeSortedContents(sortedKeys));
cfs.addSSTable(writeSortedContents(sortedKeys, context));
latch.countDown();
}
});

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

@ -52,7 +52,7 @@ import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.columniterator.IColumnIterator;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogSegment;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
@ -645,7 +645,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
assert getMemtableThreadSafe() == oldMemtable;
oldMemtable.freeze();
final CommitLogSegment.CommitLogContext ctx = writeCommitLog ? CommitLog.instance.getContext() : null;
final ReplayPosition ctx = writeCommitLog ? CommitLog.instance.getContext() : null;
// submit the memtable for any indexed sub-cfses, and our own.
List<ColumnFamilyStore> icc = new ArrayList<ColumnFamilyStore>(indexedColumns.size());
@ -657,7 +657,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
final CountDownLatch latch = new CountDownLatch(icc.size());
for (ColumnFamilyStore cfs : icc)
submitFlush(cfs.data.switchMemtable(), latch);
submitFlush(cfs.data.switchMemtable(), latch, ctx);
// we marked our memtable as frozen as part of the concurrency control,
// so even if there was nothing to flush we need to switch it out
@ -739,7 +739,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (binaryMemtable.get().isClean())
return;
submitFlush(binaryMemtable.get(), new CountDownLatch(1));
submitFlush(binaryMemtable.get(), new CountDownLatch(1), null);
}
public void updateRowCache(DecoratedKey key, ColumnFamily columnFamily)
@ -1006,10 +1006,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* flushing thread finishes sorting, which will almost always be longer than any of the flushSorter threads proper
* (since, by definition, it started last).
*/
void submitFlush(IFlushable flushable, CountDownLatch latch)
void submitFlush(IFlushable flushable, CountDownLatch latch, ReplayPosition context)
{
logger.info("Enqueuing flush of {}", flushable);
flushable.flushAndSignal(latch, flushSorter, flushWriter);
flushable.flushAndSignal(latch, flushSorter, flushWriter, context);
}
public long getMemtableColumnsCount()
@ -2116,14 +2116,15 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return intern(name);
}
public SSTableWriter createFlushWriter(long estimatedRows, long estimatedSize) throws IOException
public SSTableWriter createFlushWriter(long estimatedRows, long estimatedSize, ReplayPosition context) throws IOException
{
return new SSTableWriter(getFlushPath(estimatedSize, Descriptor.CURRENT_VERSION), estimatedRows, metadata, partitioner);
return new SSTableWriter(getFlushPath(estimatedSize, Descriptor.CURRENT_VERSION), estimatedRows, metadata, partitioner, context);
}
public SSTableWriter createCompactionWriter(long estimatedRows, String location) throws IOException
public SSTableWriter createCompactionWriter(long estimatedRows, String location, Collection<SSTableReader> sstables) throws IOException
{
return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner);
ReplayPosition rp = ReplayPosition.getReplayPosition(sstables);
return new SSTableWriter(getTempSSTablePath(location), estimatedRows, metadata, partitioner, rp);
}
public Iterable<ColumnFamilyStore> concatWithIndexes()

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

@ -564,7 +564,7 @@ public class CompactionManager implements CompactionManagerMBean
return 0;
}
writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation);
writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, sstables);
while (nni.hasNext())
{
AbstractCompactedRow row = nni.next();
@ -652,7 +652,7 @@ public class CompactionManager implements CompactionManagerMBean
assert firstRowPositionFromIndex == 0 : firstRowPositionFromIndex;
}
SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null);
SSTableWriter writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, null, Collections.singletonList(sstable));
executor.beginCompaction(new ScrubInfo(dataFile, sstable));
int goodRows = 0, badRows = 0, emptyRows = 0;
@ -840,7 +840,7 @@ public class CompactionManager implements CompactionManagerMBean
SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next();
if (Range.isTokenInRanges(row.getKey().token, ranges))
{
writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer);
writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer, Collections.singletonList(sstable));
writer.append(getCompactedRow(row, sstable.descriptor, false));
totalkeysWritten++;
}
@ -921,13 +921,13 @@ public class CompactionManager implements CompactionManagerMBean
: new PrecompactedRow(CompactionController.getBasicController(forceDeserialize), Arrays.asList(row));
}
private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer)
private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer, Collection<SSTableReader> sstables)
throws IOException
{
if (writer == null)
{
FileUtils.createDirectory(compactionFileLocation);
writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation);
writer = cfs.createCompactionWriter(expectedBloomFilterSize, compactionFileLocation, sstables);
}
return writer;
}

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

@ -24,7 +24,9 @@ package org.apache.cassandra.db;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import org.apache.cassandra.db.commitlog.ReplayPosition;
public interface IFlushable
{
public void flushAndSignal(CountDownLatch condition, ExecutorService sorter, ExecutorService writer);
public void flushAndSignal(CountDownLatch condition, ExecutorService sorter, ExecutorService writer, ReplayPosition context);
}

9
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.db.columniterator.IColumnIterator;
import org.apache.cassandra.db.columniterator.SimpleAbstractColumnIterator;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.AbstractColumnIterator;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.SliceQueryFilter;
@ -231,7 +232,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
}
private SSTableReader writeSortedContents() throws IOException
private SSTableReader writeSortedContents(ReplayPosition context) throws IOException
{
logger.info("Writing " + this);
@ -242,7 +243,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
+ keySize // keys in data file
+ currentThroughput.get()) // data
* 1.2); // bloom filter and row index overhead
SSTableWriter writer = cfs.createFlushWriter(columnFamilies.size(), estimatedSize);
SSTableWriter writer = cfs.createFlushWriter(columnFamilies.size(), estimatedSize, context);
// (we can't clear out the map as-we-go to free up memory,
// since the memtable is being used for queries in the "pending flush" category)
@ -255,7 +256,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
return ssTable;
}
public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer)
public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context)
{
writer.execute(new WrappedRunnable()
{
@ -266,7 +267,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
{
if (!cfs.isDropped())
{
SSTableReader sstable = writeSortedContents();
SSTableReader sstable = writeSortedContents(context);
cfs.replaceFlushed(Memtable.this, sstable);
}
}

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

@ -36,6 +36,7 @@ import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.LocalToken;
@ -748,7 +749,8 @@ public class Table
}
@Override
public String toString() {
public String toString()
{
return getClass().getSimpleName() + "(name='" + name + "')";
}
}

106
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -27,6 +27,11 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
import com.google.common.collect.Iterables;
import com.google.common.collect.Ordering;
import org.apache.cassandra.db.*;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.net.MessagingService;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
@ -37,10 +42,6 @@ import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.UnserializableColumnFamilyException;
import org.apache.cassandra.io.DeletionService;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileUtils;
@ -133,7 +134,7 @@ public class CommitLog
return false;
}
public static void recover() throws IOException
public static int recover() throws IOException
{
String directory = DatabaseDescriptor.getCommitLogLocation();
File[] files = new File(directory).listFiles(new FilenameFilter()
@ -149,49 +150,63 @@ public class CommitLog
if (files.length == 0)
{
logger.info("No commitlog files found; skipping replay");
return;
return 0;
}
Arrays.sort(files, new FileUtils.FileComparator());
logger.info("Replaying " + StringUtils.join(files, ", "));
recover(files);
int replayed = recover(files);
for (File f : files)
{
FileUtils.delete(CommitLogHeader.getHeaderPathFromSegmentPath(f.getAbsolutePath())); // may not actually exist
if (!f.delete())
logger.error("Unable to remove " + f + "; you should remove it manually or next restart will replay it again (harmless, but time-consuming)");
}
logger.info("Log replay complete");
logger.info("Log replay complete, " + replayed + " replayed mutations");
return replayed;
}
public static void recover(File[] clogs) throws IOException
// returns the number of replayed mutation (useful for tests in particular)
public static int recover(File[] clogs) throws IOException
{
Set<Table> tablesRecovered = new HashSet<Table>();
List<Future<?>> futures = new ArrayList<Future<?>>();
byte[] bytes = new byte[4096];
Map<Integer, AtomicInteger> invalidMutations = new HashMap<Integer, AtomicInteger>();
for (File file : clogs)
// count the number of replayed mutation. We don't really care about atomicity, but we need it to be a reference.
final AtomicInteger replayedCount = new AtomicInteger();
// compute per-CF and global replay positions
final Map<Integer, ReplayPosition> cfPositions = new HashMap<Integer, ReplayPosition>();
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// it's important to call RP.gRP per-cf, before aggregating all the positions w/ the Ordering.min call
// below: gRP will return NONE if there are no flushed sstables, which is important to have in the
// list (otherwise we'll just start replay from the first flush position that we do have, which is not correct).
ReplayPosition rp = ReplayPosition.getReplayPosition(cfs.getSSTables());
cfPositions.put(cfs.metadata.cfId, rp);
}
final ReplayPosition globalPosition = Ordering.from(ReplayPosition.comparator).min(cfPositions.values());
for (final File file : clogs)
{
final long segment = CommitLogSegment.idFromFilename(file.getName());
int bufferSize = (int) Math.min(Math.max(file.length(), 1), 32 * 1024 * 1024);
BufferedRandomAccessFile reader = new BufferedRandomAccessFile(new File(file.getAbsolutePath()), "r", bufferSize, true);
assert reader.length() <= Integer.MAX_VALUE;
try
{
CommitLogHeader clHeader = null;
int replayPosition = 0;
String headerPath = CommitLogHeader.getHeaderPathFromSegmentPath(file.getAbsolutePath());
try
{
clHeader = CommitLogHeader.readCommitLogHeader(headerPath);
replayPosition = clHeader.getReplayPosition();
}
catch (IOException ioe)
{
logger.info(headerPath + " incomplete, missing or corrupt. Everything is ok, don't panic. CommitLog will be replayed from the beginning");
logger.debug("exception was", ioe);
}
if (replayPosition < 0 || replayPosition > reader.length())
int replayPosition;
if (globalPosition.segment < segment)
replayPosition = 0;
else if (globalPosition.segment == segment)
replayPosition = globalPosition.position;
else
replayPosition = (int) reader.length();
if (replayPosition < 0 || replayPosition >= reader.length())
{
// replayPosition > reader.length() can happen if some data gets flushed before it is written to the commitlog
// (see https://issues.apache.org/jira/browse/CASSANDRA-2285)
@ -277,7 +292,6 @@ public class CommitLog
tablesRecovered.add(table);
final Collection<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(rm.getColumnFamilies());
final long entryLocation = reader.getFilePointer();
final CommitLogHeader finalHeader = clHeader;
final RowMutation frm = rm;
Runnable runnable = new WrappedRunnable()
{
@ -294,8 +308,15 @@ public class CommitLog
// null means the cf has been dropped
continue;
if (finalHeader == null || (finalHeader.isDirty(columnFamily.id()) && entryLocation > finalHeader.getPosition(columnFamily.id())))
ReplayPosition rp = cfPositions.get(columnFamily.id());
// replay if current segment is newer than last flushed one or, if it is the last known
// segment, if we are after the replay position
if (segment > rp.segment || (segment == rp.segment && entryLocation > rp.position))
{
newRm.add(columnFamily);
replayedCount.incrementAndGet();
}
}
if (!newRm.isEmpty())
{
@ -330,6 +351,8 @@ public class CommitLog
for (Table table : tablesRecovered)
futures.addAll(table.flush());
FBUtilities.waitOnFutures(futures);
return replayedCount.get();
}
private CommitLogSegment currentSegment()
@ -337,11 +360,11 @@ public class CommitLog
return segments.getLast();
}
public CommitLogSegment.CommitLogContext getContext()
public ReplayPosition getContext()
{
Callable<CommitLogSegment.CommitLogContext> task = new Callable<CommitLogSegment.CommitLogContext>()
Callable<ReplayPosition> task = new Callable<ReplayPosition>()
{
public CommitLogSegment.CommitLogContext call() throws Exception
public ReplayPosition call() throws Exception
{
return currentSegment().getContext();
}
@ -377,7 +400,7 @@ public class CommitLog
* The bit flag associated with this column family is set in the
* header and this is used to decide if the log file can be deleted.
*/
public void discardCompletedSegments(final Integer cfId, final CommitLogSegment.CommitLogContext context) throws IOException
public void discardCompletedSegments(final Integer cfId, final ReplayPosition context) throws IOException
{
Callable task = new Callable()
{
@ -408,7 +431,7 @@ public class CommitLog
* param @ id id of the columnFamily being flushed to disk.
*
*/
private void discardCompletedSegmentsInternal(CommitLogSegment.CommitLogContext context, Integer id) throws IOException
private void discardCompletedSegmentsInternal(ReplayPosition context, Integer id) throws IOException
{
if (logger.isDebugEnabled())
logger.debug("discard completed log segments for " + context + ", column family " + id + ".");
@ -423,26 +446,20 @@ public class CommitLog
while (iter.hasNext())
{
CommitLogSegment segment = iter.next();
CommitLogHeader header = segment.getHeader();
if (segment.equals(context.getSegment()))
if (segment.id == context.segment)
{
// we can't just mark the segment where the flush happened clean,
// since there may have been writes to it between when the flush
// started and when it finished. so mark the flush position as
// the replay point for this CF, instead.
if (logger.isDebugEnabled())
logger.debug("Marking replay position " + context.position + " on commit log " + segment);
header.turnOn(id, context.position);
segment.writeHeader();
// started and when it finished.
segment.turnOn(id);
break;
}
header.turnOff(id);
if (header.isSafeToDelete() && iter.hasNext())
segment.turnOff(id);
if (segment.isSafeToDelete() && iter.hasNext())
{
logger.info("Discarding obsolete commit log:" + segment);
segment.close();
DeletionService.executeDelete(segment.getHeaderPath());
DeletionService.executeDelete(segment.getPath());
// usually this will be the first (remaining) segment, but not always, if segment A contains
// writes to a CF that is unflushed but is followed by segment B whose CFs are all flushed.
@ -451,8 +468,7 @@ public class CommitLog
else
{
if (logger.isDebugEnabled())
logger.debug("Not safe to delete commit log " + segment + "; dirty is " + header.dirtyString());
segment.writeHeader();
logger.debug("Not safe to delete commit log " + segment + "; dirty is " + segment.dirtyString() + "; hasNext: " + iter.hasNext());
}
}
}

198
src/java/org/apache/cassandra/db/commitlog/CommitLogHeader.java
View File

@ -1,198 +0,0 @@
/**
* 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.db.commitlog;
import java.io.*;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
import org.apache.cassandra.io.ICompactSerializer2;
import org.apache.cassandra.io.util.FileUtils;
public class CommitLogHeader
{
public static String getHeaderPathFromSegment(CommitLogSegment segment)
{
return getHeaderPathFromSegmentPath(segment.getPath());
}
public static String getHeaderPathFromSegmentPath(String segmentPath)
{
return segmentPath + ".header";
}
public static CommitLogHeaderSerializer serializer = new CommitLogHeaderSerializer();
private Map<Integer, Integer> cfDirtiedAt; // position at which each CF was last flushed
CommitLogHeader()
{
this(new HashMap<Integer, Integer>());
}
/*
* This ctor is used while deserializing. This ctor
* also builds an index of position to column family
* Id.
*/
private CommitLogHeader(Map<Integer, Integer> cfDirtiedAt)
{
this.cfDirtiedAt = cfDirtiedAt;
}
boolean isDirty(Integer cfId)
{
return cfDirtiedAt.containsKey(cfId);
}
int getPosition(Integer cfId)
{
Integer x = cfDirtiedAt.get(cfId);
return x == null ? 0 : x;
}
void turnOn(Integer cfId, long position)
{
assert position >= 0 && position <= Integer.MAX_VALUE;
cfDirtiedAt.put(cfId, (int)position);
}
void turnOff(Integer cfId)
{
cfDirtiedAt.remove(cfId);
}
boolean isSafeToDelete() throws IOException
{
return cfDirtiedAt.isEmpty();
}
// we use cf ids. getting the cf names would be pretty pretty expensive.
public String toString()
{
StringBuilder sb = new StringBuilder("");
sb.append("CLH(dirty+flushed={");
for (Map.Entry<Integer, Integer> entry : cfDirtiedAt.entrySet())
{
sb.append(entry.getKey()).append(": ").append(entry.getValue()).append(", ");
}
sb.append("})");
return sb.toString();
}
public String dirtyString()
{
StringBuilder sb = new StringBuilder();
for (Map.Entry<Integer, Integer> entry : cfDirtiedAt.entrySet())
sb.append(entry.getKey()).append(", ");
return sb.toString();
}
static void writeCommitLogHeader(CommitLogHeader header, String headerFile) throws IOException
{
DataOutputStream out = null;
try
{
/*
* FileOutputStream doesn't sync on flush/close.
* As headers are "optional" now there is no reason to sync it.
* This provides nearly double the performance of BRAF, more under heavey load.
*/
out = new DataOutputStream(new FileOutputStream(headerFile));
serializer.serialize(header, out);
}
finally
{
if (out != null)
out.close();
}
}
static CommitLogHeader readCommitLogHeader(String headerFile) throws IOException
{
DataInputStream reader = null;
try
{
reader = new DataInputStream(new BufferedInputStream(new FileInputStream(headerFile)));
return serializer.deserialize(reader);
}
finally
{
FileUtils.closeQuietly(reader);
}
}
int getReplayPosition()
{
return cfDirtiedAt.isEmpty() ? -1 : Collections.min(cfDirtiedAt.values());
}
static class CommitLogHeaderSerializer implements ICompactSerializer2<CommitLogHeader>
{
public void serialize(CommitLogHeader clHeader, DataOutput dos) throws IOException
{
Checksum checksum = new CRC32();
// write the first checksum after the fixed-size part, so we won't read garbage lastFlushedAt data.
dos.writeInt(clHeader.cfDirtiedAt.size()); // 4
checksum.update(clHeader.cfDirtiedAt.size());
dos.writeLong(checksum.getValue());
// write the 2nd checksum after the lastflushedat map
for (Map.Entry<Integer, Integer> entry : clHeader.cfDirtiedAt.entrySet())
{
dos.writeInt(entry.getKey()); // 4
checksum.update(entry.getKey());
dos.writeInt(entry.getValue()); // 4
checksum.update(entry.getValue());
}
dos.writeLong(checksum.getValue());
}
public CommitLogHeader deserialize(DataInput dis) throws IOException
{
Checksum checksum = new CRC32();
int lastFlushedAtSize = dis.readInt();
checksum.update(lastFlushedAtSize);
if (checksum.getValue() != dis.readLong())
{
throw new IOException("Invalid or corrupt commitlog header");
}
Map<Integer, Integer> lastFlushedAt = new HashMap<Integer, Integer>();
for (int i = 0; i < lastFlushedAtSize; i++)
{
int key = dis.readInt();
checksum.update(key);
int value = dis.readInt();
checksum.update(value);
lastFlushedAt.put(key, value);
}
if (checksum.getValue() != dis.readLong())
{
throw new IOException("Invalid or corrupt commitlog header");
}
return new CommitLogHeader(lastFlushedAt);
}
}
}

128
src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
View File

@ -23,6 +23,10 @@ package org.apache.cassandra.db.commitlog;
import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
@ -30,30 +34,30 @@ import org.apache.cassandra.net.MessagingService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
public class CommitLogSegment
{
private static final Logger logger = LoggerFactory.getLogger(CommitLogSegment.class);
private static Pattern COMMIT_LOG_FILE_PATTERN = Pattern.compile("CommitLog-(\\d+).log");
public final long id;
private final BufferedRandomAccessFile logWriter;
private final CommitLogHeader header;
// cache which cf is dirty in this segment to avoid having to lookup all ReplayPositions to decide if we could delete this segment
private Set<Integer> cfDirty = new HashSet<Integer>();
public CommitLogSegment()
{
this.header = new CommitLogHeader();
String logFile = DatabaseDescriptor.getCommitLogLocation() + File.separator + "CommitLog-" + System.currentTimeMillis() + ".log";
id = System.currentTimeMillis();
String logFile = DatabaseDescriptor.getCommitLogLocation() + File.separator + "CommitLog-" + id + ".log";
logger.info("Creating new commitlog segment " + logFile);
try
{
logWriter = createWriter(logFile);
writeHeader();
}
catch (IOException e)
{
@ -61,14 +65,26 @@ public class CommitLogSegment
}
}
public static boolean possibleCommitLogFile(String filename)
// assume filename is a 'possibleCommitLogFile()'
public static long idFromFilename(String filename)
{
return filename.matches("CommitLog-\\d+.log");
Matcher matcher = COMMIT_LOG_FILE_PATTERN.matcher(filename);
try
{
if (matcher.matches())
return Long.valueOf(matcher.group(1));
else
return -1L;
}
catch (NumberFormatException e)
{
return -1L;
}
}
public void writeHeader() throws IOException
public static boolean possibleCommitLogFile(String filename)
{
CommitLogHeader.writeCommitLogHeader(header, getHeaderPath());
return COMMIT_LOG_FILE_PATTERN.matcher(filename).matches();
}
private static BufferedRandomAccessFile createWriter(String file) throws IOException
@ -76,35 +92,14 @@ public class CommitLogSegment
return new BufferedRandomAccessFile(new File(file), "rw", 128 * 1024, true);
}
public CommitLogSegment.CommitLogContext write(RowMutation rowMutation) throws IOException
public ReplayPosition write(RowMutation rowMutation) throws IOException
{
long currentPosition = -1L;
try
{
currentPosition = logWriter.getFilePointer();
CommitLogSegment.CommitLogContext cLogCtx = new CommitLogSegment.CommitLogContext(currentPosition);
// update header
for (ColumnFamily columnFamily : rowMutation.getColumnFamilies())
{
// we can ignore the serialized map in the header (and avoid deserializing it) since we know we are
// writing the cfs as they exist now. check for null cfm in case a cl write goes through after the cf is
// defined but before a new segment is created.
CFMetaData cfm = DatabaseDescriptor.getCFMetaData(columnFamily.id());
if (cfm == null)
{
logger.error("Attempted to write commit log entry for unrecognized column family: " + columnFamily.id());
}
else
{
Integer id = cfm.cfId;
if (!header.isDirty(id))
{
header.turnOn(id, logWriter.getFilePointer());
writeHeader();
}
}
}
assert currentPosition <= Integer.MAX_VALUE;
ReplayPosition cLogCtx = new ReplayPosition(id, (int) currentPosition);
// write mutation, w/ checksum on the size and data
Checksum checksum = new CRC32();
@ -131,14 +126,11 @@ public class CommitLogSegment
logWriter.sync();
}
public CommitLogContext getContext()
public ReplayPosition getContext()
{
return new CommitLogContext(logWriter.getFilePointer());
}
public CommitLogHeader getHeader()
{
return header;
long position = logWriter.getFilePointer();
assert position <= Integer.MAX_VALUE;
return new ReplayPosition(id, (int) position);
}
public String getPath()
@ -146,9 +138,9 @@ public class CommitLogSegment
return logWriter.getPath();
}
public String getHeaderPath()
public String getName()
{
return CommitLogHeader.getHeaderPathFromSegment(this);
return logWriter.getPath().substring(logWriter.getPath().lastIndexOf(File.separator) + 1);
}
public long length()
@ -175,34 +167,34 @@ public class CommitLogSegment
}
}
void turnOn(Integer cfId)
{
cfDirty.add(cfId);
}
void turnOff(Integer cfId)
{
cfDirty.remove(cfId);
}
// For debugging, not fast
String dirtyString()
{
StringBuilder sb = new StringBuilder();
for (Integer cfId : cfDirty)
sb.append(DatabaseDescriptor.getCFMetaData(cfId).cfName).append(" (").append(cfId).append("), ");
return sb.toString();
}
boolean isSafeToDelete()
{
return cfDirty.isEmpty();
}
@Override
public String toString()
{
return "CommitLogSegment(" + logWriter.getPath() + ')';
}
public class CommitLogContext
{
public final long position;
public CommitLogContext(long position)
{
assert position >= 0;
this.position = position;
}
public CommitLogSegment getSegment()
{
return CommitLogSegment.this;
}
@Override
public String toString()
{
return "CommitLogContext(" +
"file='" + logWriter.getPath() + '\'' +
", position=" + position +
')';
}
}
}

115
src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java Normal file
View File

@ -0,0 +1,115 @@
package org.apache.cassandra.db.commitlog;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Comparator;
import com.google.common.base.Function;
import com.google.common.collect.Iterables;
import com.google.common.collect.Ordering;
import org.apache.cassandra.io.ICompactSerializer2;
import org.apache.cassandra.io.sstable.SSTable;
public class ReplayPosition implements Comparable<ReplayPosition>
{
public static final ReplayPositionSerializer serializer = new ReplayPositionSerializer();
// NONE is used for SSTables that are streamed from other nodes and thus have no relationship
// with our local commitlog. The values satisfy the critera that
// - no real commitlog segment will have the given id
// - it will sort before any real replayposition, so it will be effectively ignored by getReplayPosition
public static final ReplayPosition NONE = new ReplayPosition(-1, 0);
/**
* Convenience method to compute the replay position for a group of SSTables.
* @param sstables
* @return the most recent (highest) replay position
*/
public static ReplayPosition getReplayPosition(Iterable<? extends SSTable> sstables)
{
if (Iterables.isEmpty(sstables))
return NONE;
Function<SSTable, ReplayPosition> f = new Function<SSTable, ReplayPosition>()
{
public ReplayPosition apply(SSTable sstable)
{
return sstable.replayPosition;
}
};
Ordering<ReplayPosition> ordering = Ordering.from(ReplayPosition.comparator);
return ordering.max(Iterables.transform(sstables, f));
}
public final long segment;
public final int position;
public static final Comparator<ReplayPosition> comparator = new Comparator<ReplayPosition>()
{
public int compare(ReplayPosition o1, ReplayPosition o2)
{
if (o1.segment != o2.segment)
return new Long(o1.segment).compareTo(o2.segment);
return new Integer(o1.position).compareTo(o2.position);
}
};
public ReplayPosition(long segment, int position)
{
this.segment = segment;
assert position >= 0;
this.position = position;
}
public int compareTo(ReplayPosition other)
{
return comparator.compare(this, other);
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ReplayPosition that = (ReplayPosition) o;
if (position != that.position) return false;
return segment == that.segment;
}
@Override
public int hashCode()
{
int result = (int) (segment ^ (segment >>> 32));
result = 31 * result + position;
return result;
}
@Override
public String toString()
{
return "ReplayPosition(" +
"segmentId=" + segment +
", position=" + position +
')';
}
public static class ReplayPositionSerializer implements ICompactSerializer2<ReplayPosition>
{
public void serialize(ReplayPosition rp, DataOutput dos) throws IOException
{
dos.writeLong(rp.segment);
dos.writeInt(rp.position);
}
public ReplayPosition deserialize(DataInput dis) throws IOException
{
return new ReplayPosition(dis.readLong(), dis.readInt());
}
}
}

36
src/java/org/apache/cassandra/db/marshal/AbstractCommutativeType.java
View File

@ -21,6 +21,7 @@
package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.sql.Types;
import org.apache.cassandra.db.Column;
import org.apache.cassandra.db.context.CounterContext;
@ -52,4 +53,39 @@ public abstract class AbstractCommutativeType extends AbstractType<Long>
{
return Long.class;
}
public boolean isSigned()
{
return true;
}
public boolean isCaseSensitive()
{
return false;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(Long obj)
{
return obj.toString().length();
}
public int getScale(Long obj)
{
return 0;
}
public int getJdbcType()
{
return Types.INTEGER;
}
public boolean needsQuotes()
{
return false;
}
}

12
src/java/org/apache/cassandra/db/marshal/AbstractType.java
View File

@ -147,4 +147,16 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
/** returns the class this AbstractType represents. */
public abstract Class<T> getType();
//
// JDBC metadata
//
public abstract boolean isSigned();
public abstract boolean isCaseSensitive();
public abstract boolean isCurrency();
public abstract int getPrecision(T obj);
public abstract int getScale(T obj);
public abstract int getJdbcType();
public abstract boolean needsQuotes();
}

47
src/java/org/apache/cassandra/db/marshal/AbstractUUIDType.java Normal file
View File

@ -0,0 +1,47 @@
package org.apache.cassandra.db.marshal;
import java.sql.Types;
import java.util.UUID;
public abstract class AbstractUUIDType extends AbstractType<UUID>
{
public Class<UUID> getType()
{
return UUID.class;
}
public boolean isSigned()
{
return false;
}
public boolean isCaseSensitive()
{
return false;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(UUID obj)
{
return -1;
}
public int getScale(UUID obj)
{
return -1;
}
public int getJdbcType()
{
return Types.OTHER;
}
public boolean needsQuotes()
{
return false;
}
}

36
src/java/org/apache/cassandra/db/marshal/AsciiType.java
View File

@ -23,6 +23,7 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.sql.Types;
import com.google.common.base.Charsets;
@ -86,4 +87,39 @@ public class AsciiType extends AbstractType<String>
{
return String.class;
}
public boolean isSigned()
{
return false;
}
public boolean isCaseSensitive()
{
return true;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(String obj)
{
return -1;
}
public int getScale(String obj)
{
return -1;
}
public int getJdbcType()
{
return Types.VARCHAR;
}
public boolean needsQuotes()
{
return true;
}
}

36
src/java/org/apache/cassandra/db/marshal/BytesType.java
View File

@ -22,6 +22,7 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.sql.Types;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
@ -88,4 +89,39 @@ public class BytesType extends AbstractType<ByteBuffer>
{
return ByteBuffer.class;
}
public boolean isSigned()
{
return false;
}
public boolean isCaseSensitive()
{
return false;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(ByteBuffer obj)
{
return -1;
}
public int getScale(ByteBuffer obj)
{
return -1;
}
public int getJdbcType()
{
return Types.BINARY;
}
public boolean needsQuotes()
{
return true;
}
}

36
src/java/org/apache/cassandra/db/marshal/IntegerType.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.db.marshal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.sql.Types;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.TBaseHelper;
@ -169,4 +170,39 @@ public final class IntegerType extends AbstractType<BigInteger>
{
return BigInteger.class;
}
public boolean isSigned()
{
return true;
}
public boolean isCaseSensitive()
{
return false;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(BigInteger obj)
{
return obj.toString().length();
}
public int getScale(BigInteger obj)
{
return 0;
}
public int getJdbcType()
{
return Types.BIGINT;
}
public boolean needsQuotes()
{
return false;
}
}

2
src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java
View File

@ -27,7 +27,7 @@ import java.util.UUID;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;
public class LexicalUUIDType extends AbstractType<UUID>
public class LexicalUUIDType extends AbstractUUIDType
{
public static final LexicalUUIDType instance = new LexicalUUIDType();

35
src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java
View File

@ -77,4 +77,39 @@ public class LocalByPartionerType<T extends Token> extends AbstractType<ByteBuff
{
return ByteBuffer.class;
}
public boolean isSigned()
{
throw new UnsupportedOperationException();
}
public boolean isCaseSensitive()
{
throw new UnsupportedOperationException();
}
public boolean isCurrency()
{
throw new UnsupportedOperationException();
}
public int getPrecision(ByteBuffer obj)
{
throw new UnsupportedOperationException();
}
public int getScale(ByteBuffer obj)
{
throw new UnsupportedOperationException();
}
public int getJdbcType()
{
throw new UnsupportedOperationException();
}
public boolean needsQuotes()
{
throw new UnsupportedOperationException();
}
}

36
src/java/org/apache/cassandra/db/marshal/LongType.java
View File

@ -22,6 +22,7 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.sql.Types;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -109,4 +110,39 @@ public class LongType extends AbstractType<Long>
{
return Long.class;
}
public boolean isSigned()
{
return true;
}
public boolean isCaseSensitive()
{
return false;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(Long obj)
{
return obj.toString().length();
}
public int getScale(Long obj)
{
return 0;
}
public int getJdbcType()
{
return Types.INTEGER;
}
public boolean needsQuotes()
{
return false;
}
}

2
src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java
View File

@ -31,7 +31,7 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.commons.lang.time.DateUtils;
public class TimeUUIDType extends AbstractType<UUID>
public class TimeUUIDType extends AbstractUUIDType
{
public static final TimeUUIDType instance = new TimeUUIDType();

36
src/java/org/apache/cassandra/db/marshal/UTF8Type.java
View File

@ -22,6 +22,7 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.sql.Types;
import com.google.common.base.Charsets;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -196,4 +197,39 @@ public class UTF8Type extends AbstractType<String>
{
return String.class;
}
public boolean isSigned()
{
return false;
}
public boolean isCaseSensitive()
{
return true;
}
public boolean isCurrency()
{
return false;
}
public int getPrecision(String obj)
{
return -1;
}
public int getScale(String obj)
{
return -1;
}
public int getJdbcType()
{
return Types.VARCHAR;
}
public boolean needsQuotes()
{
return true;
}
}

2
src/java/org/apache/cassandra/db/marshal/UUIDType.java
View File

@ -43,7 +43,7 @@ import org.apache.cassandra.utils.UUIDGen;
* @see "com.fasterxml.uuid.UUIDComparator"
*
*/
public class UUIDType extends AbstractType<UUID>
public class UUIDType extends AbstractUUIDType
{
public static final UUIDType instance = new UUIDType();

7
src/java/org/apache/cassandra/io/sstable/Descriptor.java
View File

@ -38,7 +38,7 @@ import org.apache.cassandra.utils.Pair;
public class Descriptor
{
public static final String LEGACY_VERSION = "a";
public static final String CURRENT_VERSION = "f";
public static final String CURRENT_VERSION = "g";
public final File directory;
public final String version;
@ -80,6 +80,11 @@ public class Descriptor
isLatestVersion = version.compareTo(CURRENT_VERSION) == 0;
}
public boolean hasReplayPosition()
{
return version.compareTo("g") >= 0;
}
public String filenameFor(Component component)
{
return filenameFor(component.name());

46
src/java/org/apache/cassandra/io/sstable/SSTable.java
View File

@ -31,6 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
import org.apache.cassandra.io.util.FileUtils;
@ -67,17 +68,41 @@ public abstract class SSTable
public final CFMetaData metadata;
public final IPartitioner partitioner;
public final ReplayPosition replayPosition;
protected final EstimatedHistogram estimatedRowSize;
protected final EstimatedHistogram estimatedColumnCount;
protected SSTable(Descriptor descriptor, CFMetaData metadata, IPartitioner partitioner)
protected SSTable(Descriptor descriptor, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner)
{
this(descriptor, new HashSet<Component>(), metadata, partitioner);
this(descriptor, new HashSet<Component>(), metadata, replayPosition, partitioner);
}
protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, IPartitioner partitioner)
protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner)
{
this(descriptor, components, metadata, partitioner, defaultRowHistogram(), defaultColumnHistogram());
this(descriptor, components, metadata, replayPosition, partitioner, defaultRowHistogram(), defaultColumnHistogram());
}
protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts)
{
assert descriptor != null;
assert components != null;
assert metadata != null;
assert replayPosition != null;
assert partitioner != null;
assert rowSizes != null;
assert columnCounts != null;
this.descriptor = descriptor;
Set<Component> dataComponents = new HashSet<Component>(components);
for (Component component : components)
assert component.type != Component.Type.COMPACTED_MARKER;
this.components = Collections.unmodifiableSet(dataComponents);
this.metadata = metadata;
this.replayPosition = replayPosition;
this.partitioner = partitioner;
estimatedRowSize = rowSizes;
estimatedColumnCount = columnCounts;
}
static EstimatedHistogram defaultColumnHistogram()
@ -90,19 +115,6 @@ public abstract class SSTable
return new EstimatedHistogram(150);
}
protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, IPartitioner partitioner, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts)
{
this.descriptor = descriptor;
Set<Component> dataComponents = new HashSet<Component>(components);
for (Component component : components)
assert component.type != Component.Type.COMPACTED_MARKER;
this.components = Collections.unmodifiableSet(dataComponents);
this.metadata = metadata;
this.partitioner = partitioner;
estimatedRowSize = rowSizes;
estimatedColumnCount = columnCounts;
}
public EstimatedHistogram getEstimatedRowSize()
{
return estimatedRowSize;

15
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -34,6 +34,7 @@ import org.apache.cassandra.cache.InstrumentingCache;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.AbstractBounds;
@ -156,15 +157,18 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
EstimatedHistogram rowSizes;
EstimatedHistogram columnCounts;
File statsFile = new File(descriptor.filenameFor(SSTable.COMPONENT_STATS));
ReplayPosition rp = ReplayPosition.NONE;
if (statsFile.exists())
{
DataInputStream dis = null;
try
{
logger.debug("Load statistics for {}", descriptor);
logger.debug("Load metadata for {}", descriptor);
dis = new DataInputStream(new BufferedInputStream(new FileInputStream(statsFile)));
rowSizes = EstimatedHistogram.serializer.deserialize(dis);
columnCounts = EstimatedHistogram.serializer.deserialize(dis);
if (descriptor.hasReplayPosition())
rp = ReplayPosition.serializer.deserialize(dis);
}
finally
{
@ -178,7 +182,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
columnCounts = SSTable.defaultColumnHistogram();
}
SSTableReader sstable = new SSTableReader(descriptor, components, metadata, partitioner, null, null, null, null, System.currentTimeMillis(), rowSizes, columnCounts);
SSTableReader sstable = new SSTableReader(descriptor, components, metadata, rp, partitioner, null, null, null, null, System.currentTimeMillis(), rowSizes, columnCounts);
sstable.setTrackedBy(tracker);
// versions before 'c' encoded keys as utf-16 before hashing to the filter
@ -203,16 +207,17 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
/**
* Open a RowIndexedReader which already has its state initialized (by SSTableWriter).
*/
static SSTableReader internalOpen(Descriptor desc, Set<Component> components, CFMetaData metadata, IPartitioner partitioner, SegmentedFile ifile, SegmentedFile dfile, IndexSummary isummary, Filter bf, long maxDataAge, EstimatedHistogram rowsize,
static SSTableReader internalOpen(Descriptor desc, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner, SegmentedFile ifile, SegmentedFile dfile, IndexSummary isummary, Filter bf, long maxDataAge, EstimatedHistogram rowsize,
EstimatedHistogram columncount) throws IOException
{
assert desc != null && partitioner != null && ifile != null && dfile != null && isummary != null && bf != null;
return new SSTableReader(desc, components, metadata, partitioner, ifile, dfile, isummary, bf, maxDataAge, rowsize, columncount);
return new SSTableReader(desc, components, metadata, replayPosition, partitioner, ifile, dfile, isummary, bf, maxDataAge, rowsize, columncount);
}
private SSTableReader(Descriptor desc,
Set<Component> components,
CFMetaData metadata,
ReplayPosition replayPosition,
IPartitioner partitioner,
SegmentedFile ifile,
SegmentedFile dfile,
@ -223,7 +228,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
EstimatedHistogram columnCounts)
throws IOException
{
super(desc, components, metadata, partitioner, rowSizes, columnCounts);
super(desc, components, metadata, replayPosition, partitioner, rowSizes, columnCounts);
this.maxDataAge = maxDataAge;
this.ifile = ifile;

19
src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
View File

@ -28,6 +28,7 @@ import java.util.Set;
import com.google.common.collect.Sets;
import org.apache.cassandra.db.commitlog.ReplayPosition;
import org.apache.cassandra.io.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.slf4j.Logger;
@ -62,14 +63,15 @@ public class SSTableWriter extends SSTable
public SSTableWriter(String filename, long keyCount) throws IOException
{
this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner());
this(filename, keyCount, DatabaseDescriptor.getCFMetaData(Descriptor.fromFilename(filename)), StorageService.getPartitioner(), ReplayPosition.NONE);
}
public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner) throws IOException
public SSTableWriter(String filename, long keyCount, CFMetaData metadata, IPartitioner partitioner, ReplayPosition replayPosition) throws IOException
{
super(Descriptor.fromFilename(filename),
new HashSet<Component>(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS)),
metadata,
replayPosition,
partitioner,
SSTable.defaultRowHistogram(),
SSTable.defaultColumnHistogram());
@ -182,7 +184,7 @@ public class SSTableWriter extends SSTable
FileUtils.truncate(dataFile.getPath(), position);
// write sstable statistics
writeStatistics(descriptor, estimatedRowSize, estimatedColumnCount);
writeMetadata(descriptor, estimatedRowSize, estimatedColumnCount, replayPosition);
// remove the 'tmp' marker from all components
final Descriptor newdesc = rename(descriptor, components);
@ -190,20 +192,21 @@ public class SSTableWriter extends SSTable
// finalize in-memory state for the reader
SegmentedFile ifile = iwriter.builder.complete(newdesc.filenameFor(SSTable.COMPONENT_INDEX));
SegmentedFile dfile = dbuilder.complete(newdesc.filenameFor(SSTable.COMPONENT_DATA));
SSTableReader sstable = SSTableReader.internalOpen(newdesc, components, metadata, partitioner, ifile, dfile, iwriter.summary, iwriter.bf, maxDataAge, estimatedRowSize, estimatedColumnCount);
SSTableReader sstable = SSTableReader.internalOpen(newdesc, components, metadata, replayPosition, partitioner, ifile, dfile, iwriter.summary, iwriter.bf, maxDataAge, estimatedRowSize, estimatedColumnCount);
iwriter = null;
dbuilder = null;
return sstable;
}
private static void writeStatistics(Descriptor desc, EstimatedHistogram rowSizes, EstimatedHistogram columnnCounts) throws IOException
private static void writeMetadata(Descriptor desc, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts, ReplayPosition rp) throws IOException
{
BufferedRandomAccessFile out = new BufferedRandomAccessFile(new File(desc.filenameFor(SSTable.COMPONENT_STATS)),
"rw",
BufferedRandomAccessFile.DEFAULT_BUFFER_SIZE,
true);
EstimatedHistogram.serializer.serialize(rowSizes, out);
EstimatedHistogram.serializer.serialize(columnnCounts, out);
EstimatedHistogram.serializer.serialize(columnCounts, out);
ReplayPosition.serializer.serialize(rp, out);
out.close();
}
@ -457,7 +460,7 @@ public class SSTableWriter extends SSTable
rows++;
}
writeStatistics(desc, rowSizes, columnCounts);
writeMetadata(desc, rowSizes, columnCounts, ReplayPosition.NONE);
return rows;
}
}
@ -527,7 +530,7 @@ public class SSTableWriter extends SSTable
rows++;
}
writeStatistics(desc, rowSizes, columnCounts);
writeMetadata(desc, rowSizes, columnCounts, ReplayPosition.NONE);
if (writerDfile.getFilePointer() != dfile.getFilePointer())
{

83
test/system/test_cql.py
View File

@ -152,29 +152,68 @@ class TestCql(ThriftTester):
def test_select_row_range(self):
"retrieve a range of rows with columns"
cursor = init()
cursor.execute("""
SELECT 4 FROM StandardLongA WHERE KEY > 'ad' AND KEY < 'ag';
""")
rows = [row[0] for row in cursor.fetchall()]
assert ['ad', 'ae', 'af', 'ag'] == rows, rows
def test_select_row_range_with_limit(self):
"retrieve a limited range of rows with columns"
cursor = init()
cursor.execute("""
SELECT 1,5,9 FROM StandardLongA WHERE KEY > 'aa'
AND KEY < 'ag' LIMIT 3
""")
assert cursor.rowcount == 3
# everything
cursor.execute("SELECT * FROM StandardLongA")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa', 'ab', 'ac', 'ad', 'ae', 'af', 'ag'] == keys, keys
cursor.execute("""
SELECT 20,40 FROM StandardIntegerA WHERE KEY > 'k1'
AND KEY < 'k7' LIMIT 5
""")
assert cursor.rowcount == 5
for i in range(5):
r = cursor.fetchone()
assert r[0] == "k%d" % (i+1)
# [start, end], mid-row
cursor.execute("SELECT * FROM StandardLongA WHERE KEY >= 'ad' AND KEY <= 'ag'")
keys = [row[0] for row in cursor.fetchall()]
assert ['ad', 'ae', 'af', 'ag'] == keys, keys
# (start, end), mid-row
cursor.execute("SELECT * FROM StandardLongA WHERE KEY > 'ad' AND KEY < 'ag'")
keys = [row[0] for row in cursor.fetchall()]
assert ['ae', 'af'] == keys, keys
# [start, end], full-row
cursor.execute("SELECT * FROM StandardLongA WHERE KEY >= 'aa' AND KEY <= 'ag'")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa', 'ab', 'ac', 'ad', 'ae', 'af', 'ag'] == keys, keys
# (start, end), full-row
cursor.execute("SELECT * FROM StandardLongA WHERE KEY > 'a' AND KEY < 'g'")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa', 'ab', 'ac', 'ad', 'ae', 'af', 'ag'] == keys, keys
# LIMIT tests
# no WHERE
cursor.execute("SELECT * FROM StandardLongA LIMIT 1")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa'] == keys, keys
# with >=, non-existing key
cursor.execute("SELECT * FROM StandardLongA WHERE KEY >= 'a' LIMIT 1")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa'] == keys, keys
# with >=, existing key
cursor.execute("SELECT * FROM StandardLongA WHERE KEY >= 'aa' LIMIT 1")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa'] == keys, keys
# with >, non-existing key
cursor.execute("SELECT * FROM StandardLongA WHERE KEY > 'a' LIMIT 1")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa'] == keys, keys
# with >, existing key
cursor.execute("SELECT * FROM StandardLongA WHERE KEY > 'aa' LIMIT 1")
keys = [row[0] for row in cursor.fetchall()]
assert ['ab'] == keys, keys
# with both > and <, existing keys
cursor.execute("SELECT * FROM StandardLongA WHERE KEY > 'aa' and KEY < 'ag' LIMIT 5")
keys = [row[0] for row in cursor.fetchall()]
assert ['ab', 'ac', 'ad', 'ae', 'af'] == keys, keys
# with both > and <, non-existing keys
cursor.execute("SELECT * FROM StandardLongA WHERE KEY > 'a' and KEY < 'b' LIMIT 5")
keys = [row[0] for row in cursor.fetchall()]
assert ['aa', 'ab', 'ac', 'ad', 'ae'] == keys, keys
def test_select_columns_slice(self):
"column slice tests"
@ -287,7 +326,7 @@ class TestCql(ThriftTester):
cursor = init()
cursor.execute("""
SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
AND KEY > 'asmithZ'
AND KEY >= 'asmithZ'
""")
assert cursor.rowcount == 1
r = cursor.fetchone()

70
test/unit/org/apache/cassandra/db/CommitLogTest.java
View File

@ -27,36 +27,15 @@ import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.commitlog.CommitLogHeader;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.utils.Pair;
public class CommitLogTest extends CleanupHelper
{
@Test
public void testRecoveryWithEmptyHeader() throws Exception
{
testRecovery(new byte[0], new byte[10]);
}
@Test
public void testRecoveryWithShortHeader() throws Exception
{
testRecovery(new byte[2], new byte[10]);
}
@Test
public void testRecoveryWithGarbageHeader() throws Exception
{
byte[] garbage = new byte[100];
(new java.util.Random()).nextBytes(garbage);
testRecovery(garbage, garbage);
}
@Test
public void testRecoveryWithEmptyLog() throws Exception
{
CommitLog.recover(new File[] {tmpFiles().right});
CommitLog.recover(new File[] {tmpFile()});
}
@Test
@ -69,13 +48,13 @@ public class CommitLogTest extends CleanupHelper
@Test
public void testRecoveryWithShortSize() throws Exception
{
testRecovery(new byte[0], new byte[2]);
testRecovery(new byte[2]);
}
@Test
public void testRecoveryWithShortCheckSum() throws Exception
{
testRecovery(new byte[0], new byte[6]);
testRecovery(new byte[6]);
}
@Test
@ -83,7 +62,7 @@ public class CommitLogTest extends CleanupHelper
{
byte[] garbage = new byte[100];
(new java.util.Random()).nextBytes(garbage);
testRecovery(new byte[0], garbage);
testRecovery(garbage);
}
@Test
@ -108,30 +87,6 @@ public class CommitLogTest extends CleanupHelper
testRecoveryWithBadSizeArgument(-10, 10); // negative size, but no EOF
}
@Test
public void testRecoveryWithHeaderPositionGreaterThanLogLength() throws Exception
{
// Note: this can actually happen (in periodic mode) when data is flushed
// before it had time to hit the commitlog (since the header is flushed by the system)
// see https://issues.apache.org/jira/browse/CASSANDRA-2285
ByteArrayOutputStream out = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(out);
Checksum checksum = new CRC32();
// write the first checksum after the fixed-size part, so we won't read garbage lastFlushedAt data.
dos.writeInt(1);
checksum.update(1);
dos.writeLong(checksum.getValue());
dos.writeInt(0);
checksum.update(0);
dos.writeInt(200);
checksum.update(200);
dos.writeLong(checksum.getValue());
dos.close();
testRecovery(out.toByteArray(), new byte[0]);
}
protected void testRecoveryWithBadSizeArgument(int size, int dataSize) throws Exception
{
Checksum checksum = new CRC32();
@ -147,29 +102,22 @@ public class CommitLogTest extends CleanupHelper
dout.writeLong(checksum);
dout.write(new byte[dataSize]);
dout.close();
testRecovery(new byte[0], out.toByteArray());
testRecovery(out.toByteArray());
}
protected Pair<File, File> tmpFiles() throws IOException
protected File tmpFile() throws IOException
{
File logFile = File.createTempFile("testRecoveryWithPartiallyWrittenHeaderTestFile", null);
File headerFile = new File(CommitLogHeader.getHeaderPathFromSegmentPath(logFile.getAbsolutePath()));
logFile.deleteOnExit();
headerFile.deleteOnExit();
assert logFile.length() == 0;
assert headerFile.length() == 0;
return new Pair<File, File>(headerFile, logFile);
return logFile;
}
protected void testRecovery(byte[] headerData, byte[] logData) throws Exception
protected void testRecovery(byte[] logData) throws Exception
{
Pair<File, File> tmpFiles = tmpFiles();
File logFile = tmpFiles.right;
File headerFile = tmpFiles.left;
File logFile = tmpFile();
OutputStream lout = new FileOutputStream(logFile);
OutputStream hout = new FileOutputStream(headerFile);
lout.write(logData);
hout.write(headerData);
//statics make it annoying to test things correctly
CommitLog.recover(new File[] {logFile}); //CASSANDRA-1119 / CASSANDRA-1179 throw on failure*/
}

2
test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java
View File

@ -223,7 +223,7 @@ public class CompactionsPurgeTest extends CleanupHelper
// Check that the second insert did went in
ColumnFamily cf = cfs.getColumnFamily(QueryFilter.getIdentityFilter(key, new QueryPath(cfName)));
assert cf.getColumnCount() == 10;
assertEquals(10, cf.getColumnCount());
for (IColumn c : cf)
assert !c.isMarkedForDelete();
}

13
test/unit/org/apache/cassandra/db/RecoveryManager2Test.java
View File

@ -59,17 +59,12 @@ public class RecoveryManager2Test extends CleanupHelper
logger.debug("forcing flush");
cfs.forceBlockingFlush();
// remove Standard1 SSTable/MemTables
cfs.clearUnsafe();
logger.debug("begin manual replay");
// replay the commit log (nothing should be replayed since everything was flushed)
// replay the commit log (nothing on Standard1 should be replayed since everything was flushed, so only the row on Standard2
// will be replayed)
CommitLog.instance.resetUnsafe();
CommitLog.recover();
// since everything that was flushed was removed (i.e. clearUnsafe)
// and the commit shouldn't have replayed anything, there should be no data
assert Util.getRangeSlice(cfs).isEmpty();
int replayed = CommitLog.recover();
assert replayed == 1 : "Expecting only 1 replayed mutation, got " + replayed;
}
private void insertRow(String cfname, String key) throws IOException

50
test/unit/org/apache/cassandra/db/commitlog/CommitLogHeaderTest.java
View File

@ -1,50 +0,0 @@
/**
* 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
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* 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.db.commitlog;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
public class CommitLogHeaderTest extends SchemaLoader
{
@Test
public void testEmptyHeader()
{
CommitLogHeader clh = new CommitLogHeader();
assert clh.getReplayPosition() < 0;
}
@Test
public void lowestPositionWithZero()
{
CommitLogHeader clh = new CommitLogHeader();
clh.turnOn(2, 34);
assert clh.getReplayPosition() == 34;
clh.turnOn(100, 0);
assert clh.getReplayPosition() == 0;
clh.turnOn(65, 2);
assert clh.getReplayPosition() == 0;
}
}