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

Fix int/bigint in CassandraStorage 

Patch by Alex Liu, reviewed by brandonwilliams for CASSANDRA-6102
This commit is contained in:
Brandon Williams 2013-10-11 15:22:35 -05:00
parent eee485eb63
commit 639c01a350
3 changed files with 85 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
View File

@ -97,7 +97,7 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
protected String outputFormatClass;
protected int splitSize = 64 * 1024;
protected String partitionerClass;
protected boolean usePartitionFilter = false;
protected boolean usePartitionFilter = false;
public AbstractCassandraStorage()
{
@ -116,8 +116,9 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
}
/** convert a column to a tuple */
protected Tuple columnToTuple(IColumn col, CfDef cfDef, AbstractType comparator) throws IOException
protected Tuple columnToTuple(IColumn col, CfInfo cfInfo, AbstractType comparator) throws IOException
{
CfDef cfDef = cfInfo.cfDef;
Tuple pair = TupleFactory.getInstance().newTuple(2);
// name
@ -131,13 +132,21 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
{
// standard
Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
if (validators.get(col.name()) == null)
ByteBuffer colName;
if (cfInfo.cql3Table && !cfInfo.compactCqlTable)
{
ByteBuffer[] names = ((AbstractCompositeType) parseType(cfDef.comparator_type)).split(col.name());
colName = names[names.length-1];
}
else
colName = col.name();
if (validators.get(colName) == null)
{
Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
setTupleValue(pair, 1, cassandraToObj(marshallers.get(MarshallerType.DEFAULT_VALIDATOR), col.value()));
}
else
setTupleValue(pair, 1, cassandraToObj(validators.get(col.name()), col.value()));
setTupleValue(pair, 1, cassandraToObj(validators.get(colName), col.value()));
return pair;
}
else
@ -145,7 +154,7 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
// super
ArrayList<Tuple> subcols = new ArrayList<Tuple>();
for (IColumn subcol : col.getSubColumns())
subcols.add(columnToTuple(subcol, cfDef, parseType(cfDef.getSubcomparator_type())));
subcols.add(columnToTuple(subcol, cfInfo, parseType(cfDef.getSubcomparator_type())));
pair.set(1, new DefaultDataBag(subcols));
}
@ -168,11 +177,16 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
}
/** get the columnfamily definition for the signature */
protected CfDef getCfDef(String signature) throws IOException
protected CfInfo getCfInfo(String signature) throws IOException
{
UDFContext context = UDFContext.getUDFContext();
Properties property = context.getUDFProperties(AbstractCassandraStorage.class);
return cfdefFromString(property.getProperty(signature));
String prop = property.getProperty(signature);
CfInfo cfInfo = new CfInfo();
cfInfo.cfDef = cfdefFromString(prop.substring(2));
cfInfo.compactCqlTable = prop.charAt(0) == '1' ? true : false;
cfInfo.cql3Table = prop.charAt(1) == '1' ? true : false;
return cfInfo;
}
/** construct a map to store the mashaller type to cassandra data type mapping */
@ -329,10 +343,7 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
return DataType.INTEGER;
else if (type instanceof AsciiType ||
type instanceof UTF8Type ||
type instanceof DecimalType ||
type instanceof InetAddressType ||
type instanceof LexicalUUIDType ||
type instanceof UUIDType )
type instanceof DecimalType || type instanceof InetAddressType )
return DataType.CHARARRAY;
else if (type instanceof FloatType)
return DataType.FLOAT;
@ -513,11 +524,15 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
}
}
// compose the CfDef for the columfamily
CfDef cfDef = getCfDef(client);
// compose the CfInfo for the columfamily
CfInfo cfInfo = getCfInfo(client);
if (cfDef != null)
properties.setProperty(signature, cfdefToString(cfDef));
if (cfInfo.cfDef != null)
{
StringBuilder sb = new StringBuilder();
sb.append(cfInfo.compactCqlTable ? 1 : 0).append(cfInfo.cql3Table ? 1: 0).append(cfdefToString(cfInfo.cfDef));
properties.setProperty(signature, sb.toString());
}
else
throw new IOException(String.format("Column family '%s' not found in keyspace '%s'",
column_family,
@ -563,17 +578,17 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
return cfDef;
}
/** return the CfDef for the column family */
protected CfDef getCfDef(Cassandra.Client client)
/** return the CfInf for the column family */
protected CfInfo getCfInfo(Cassandra.Client client)
throws InvalidRequestException,
UnavailableException,
TimedOutException,
SchemaDisagreementException,
TException,
CharacterCodingException,
NotFoundException,
org.apache.cassandra.exceptions.InvalidRequestException,
ConfigurationException
ConfigurationException,
IOException
{
// get CF meta data
String query = "SELECT type," +
@ -627,12 +642,19 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
else
cql3Table = true;
}
cfDef.column_metadata = getColumnMetadata(client, cql3Table);
return cfDef;
cfDef.column_metadata = getColumnMetadata(client);
CfInfo cfInfo = new CfInfo();
cfInfo.cfDef = cfDef;
if (cql3Table && !(parseType(cfDef.comparator_type) instanceof AbstractCompositeType))
cfInfo.compactCqlTable = true;
if (cql3Table)
cfInfo.cql3Table = true;
return cfInfo;
}
/** get a list of columns */
protected abstract List<ColumnDef> getColumnMetadata(Cassandra.Client client, boolean cql3Table)
protected abstract List<ColumnDef> getColumnMetadata(Cassandra.Client client)
throws InvalidRequestException,
UnavailableException,
TimedOutException,
@ -749,7 +771,7 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
/** get a list of columns with defined index*/
protected List<ColumnDef> getIndexes() throws IOException
{
CfDef cfdef = getCfDef(loadSignature);
CfDef cfdef = getCfInfo(loadSignature).cfDef;
List<ColumnDef> indexes = new ArrayList<ColumnDef>();
for (ColumnDef cdef : cfdef.column_metadata)
{
@ -778,15 +800,17 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
protected Object cassandraToObj(AbstractType validator, ByteBuffer value)
{
if (validator instanceof DecimalType ||
validator instanceof InetAddressType ||
validator instanceof LexicalUUIDType ||
validator instanceof UUIDType)
{
if (validator instanceof DecimalType || validator instanceof InetAddressType)
return validator.getString(value);
}
else
return validator.compose(value);
}
protected class CfInfo
{
boolean compactCqlTable = false;
boolean cql3Table = false;
CfDef cfDef;
}
}

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

@ -101,7 +101,8 @@ public class CassandraStorage extends AbstractCassandraStorage
/** read wide row*/
public Tuple getNextWide() throws IOException
{
CfDef cfDef = getCfDef(loadSignature);
CfInfo cfInfo = getCfInfo(loadSignature);
CfDef cfDef = cfInfo.cfDef;
ByteBuffer key = null;
Tuple tuple = null;
DefaultDataBag bag = new DefaultDataBag();
@ -124,7 +125,7 @@ public class CassandraStorage extends AbstractCassandraStorage
}
for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
{
bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
}
lastKey = null;
lastRow = null;
@ -162,7 +163,7 @@ public class CassandraStorage extends AbstractCassandraStorage
addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
{
bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
}
tuple.append(bag);
lastKey = key;
@ -179,14 +180,14 @@ public class CassandraStorage extends AbstractCassandraStorage
{
for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
{
bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
}
lastKey = null;
lastRow = null;
}
for (Map.Entry<ByteBuffer, IColumn> entry : row.entrySet())
{
bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
}
}
}
@ -208,7 +209,8 @@ public class CassandraStorage extends AbstractCassandraStorage
if (!reader.nextKeyValue())
return null;
CfDef cfDef = getCfDef(loadSignature);
CfInfo cfInfo = getCfInfo(loadSignature);
CfDef cfDef = cfInfo.cfDef;
ByteBuffer key = reader.getCurrentKey();
Map<ByteBuffer, IColumn> cf = reader.getCurrentValue();
assert key != null && cf != null;
@ -224,11 +226,21 @@ public class CassandraStorage extends AbstractCassandraStorage
// take care to iterate these in the same order as the schema does
for (ColumnDef cdef : cfDef.column_metadata)
{
if (cf.containsKey(cdef.name))
boolean hasColumn = false;
boolean cql3Table = false;
try
{
tuple.append(columnToTuple(cf.get(cdef.name), cfDef, parseType(cfDef.getComparator_type())));
hasColumn = cf.containsKey(cdef.name);
}
else
catch (Exception e)
{
cql3Table = true;
}
if (hasColumn)
{
tuple.append(columnToTuple(cf.get(cdef.name), cfInfo, parseType(cfDef.getComparator_type())));
}
else if (!cql3Table)
{ // otherwise, we need to add an empty tuple to take its place
tuple.append(TupleFactory.getInstance().newTuple());
}
@ -238,7 +250,7 @@ public class CassandraStorage extends AbstractCassandraStorage
for (Map.Entry<ByteBuffer, IColumn> entry : cf.entrySet())
{
if (!added.containsKey(entry.getKey()))
bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
}
tuple.append(bag);
// finally, special top-level indexes if needed
@ -246,7 +258,7 @@ public class CassandraStorage extends AbstractCassandraStorage
{
for (ColumnDef cdef : getIndexes())
{
Tuple throwaway = columnToTuple(cf.get(cdef.name), cfDef, parseType(cfDef.getComparator_type()));
Tuple throwaway = columnToTuple(cf.get(cdef.name), cfInfo, parseType(cfDef.getComparator_type()));
tuple.append(throwaway.get(1));
}
}
@ -356,7 +368,8 @@ public class CassandraStorage extends AbstractCassandraStorage
public ResourceSchema getSchema(String location, Job job) throws IOException
{
setLocation(location, job);
CfDef cfDef = getCfDef(loadSignature);
CfInfo cfInfo = getCfInfo(loadSignature);
CfDef cfDef = cfInfo.cfDef;
if (cfDef.column_type.equals("Super"))
return null;
@ -403,7 +416,7 @@ public class CassandraStorage extends AbstractCassandraStorage
// add the key first, then the indexed columns, and finally the bag
allSchemaFields.add(keyFieldSchema);
if (!widerows)
if (!widerows && (cfInfo.compactCqlTable || !cfInfo.cql3Table))
{
// defined validators/indexes
for (ColumnDef cdef : cfDef.column_metadata)
@ -697,7 +710,7 @@ public class CassandraStorage extends AbstractCassandraStorage
}
/** get a list of column for the column family */
protected List<ColumnDef> getColumnMetadata(Cassandra.Client client, boolean cql3Table)
protected List<ColumnDef> getColumnMetadata(Cassandra.Client client)
throws InvalidRequestException,
UnavailableException,
TimedOutException,
@ -708,9 +721,6 @@ public class CassandraStorage extends AbstractCassandraStorage
ConfigurationException,
NotFoundException
{
if (cql3Table)
return new ArrayList<ColumnDef>();
return getColumnMeta(client, true, true);
}
@ -795,6 +805,5 @@ public class CassandraStorage extends AbstractCassandraStorage
"[&comparator=<comparator>][&split_size=<size>][&partitioner=<partitioner>]]': " + e.getMessage());
}
}
}

8
src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
View File

@ -95,7 +95,8 @@ public class CqlStorage extends AbstractCassandraStorage
if (!reader.nextKeyValue())
return null;
CfDef cfDef = getCfDef(loadSignature);
CfInfo cfInfo = getCfInfo(loadSignature);
CfDef cfDef = cfInfo.cfDef;
Map<String, ByteBuffer> keys = reader.getCurrentKey();
Map<String, ByteBuffer> columns = reader.getCurrentValue();
assert keys != null && columns != null;
@ -281,7 +282,8 @@ public class CqlStorage extends AbstractCassandraStorage
public ResourceSchema getSchema(String location, Job job) throws IOException
{
setLocation(location, job);
CfDef cfDef = getCfDef(loadSignature);
CfInfo cfInfo = getCfInfo(loadSignature);
CfDef cfDef = cfInfo.cfDef;
// top-level schema, no type
ResourceSchema schema = new ResourceSchema();
@ -429,7 +431,7 @@ public class CqlStorage extends AbstractCassandraStorage
}
/** include key columns */
protected List<ColumnDef> getColumnMetadata(Cassandra.Client client, boolean cql3Table)
protected List<ColumnDef> getColumnMetadata(Cassandra.Client client)
throws InvalidRequestException,
UnavailableException,
TimedOutException,