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

Correctly decode row keys in widerow mode. 

Patch by brandonwilliams reviewed by aleksey for CASSANDRA-5098
This commit is contained in:
Brandon Williams 2013-01-04 10:38:45 -06:00
parent e1206f3b87
commit 3034eeecbf
2 changed files with 11 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -3,6 +3,7 @@
* fix multithreaded compaction deadlock (CASSANDRA-4492)
* fix specifying and altering crc_check_chance (CASSANDRA-5053)
* Don't expire columns sooner than they should in 2ndary indexes (CASSANDRA-5079)
* Pig: correctly decode row keys in widerow mode (CASSANDRA-5098)
1.1.8

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

@ -143,7 +143,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
if (tuple.size() == 0) // lastRow is a new one
{
key = (ByteBuffer)reader.getCurrentKey();
tuple.append(new DataByteArray(key.array(), key.position()+key.arrayOffset(), key.limit()+key.arrayOffset()));
addKeyToTuple(tuple, key, cfDef, parseType(cfDef.getKey_validation_class()));
}
for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
{
@ -179,7 +179,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
key = (ByteBuffer)reader.getCurrentKey();
if (lastKey != null && !(key.equals(lastKey))) // last key only had one value
{
tuple.append(new DataByteArray(lastKey.array(), lastKey.position()+lastKey.arrayOffset(), lastKey.limit()+lastKey.arrayOffset()));
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())));
@ -189,7 +189,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
lastRow = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
return tuple;
}
tuple.append(new DataByteArray(key.array(), key.position()+key.arrayOffset(), key.limit()+key.arrayOffset()));
addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
}
SortedMap<ByteBuffer,IColumn> row = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
if (lastRow != null) // prepend what was read last time
@ -294,6 +294,12 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
private Tuple keyToTuple(ByteBuffer key, CfDef cfDef, AbstractType comparator) throws IOException
{
Tuple tuple = TupleFactory.getInstance().newTuple(1);
addKeyToTuple(tuple, key, cfDef, comparator);
return tuple;
}
private void addKeyToTuple(Tuple tuple, ByteBuffer key, CfDef cfDef, AbstractType comparator) throws IOException
{
if( comparator instanceof AbstractCompositeType )
{
setTupleValue(tuple, 0, composeComposite((AbstractCompositeType)comparator,key));
@ -302,7 +308,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
{
setTupleValue(tuple, 0, getDefaultMarshallers(cfDef).get(2).compose(key));
}
return tuple;
}
private Tuple columnToTuple(IColumn col, CfDef cfDef, AbstractType comparator) throws IOException