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

Merge branch 'cassandra-1.1' into cassandra-1.2 

Conflicts:
	CHANGES.txt
This commit is contained in:
Brandon Williams 2013-05-21 11:10:47 -05:00
parent b69c1aa4cc 7d2ce5f957
commit 9d0eec2171
3 changed files with 13 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -1,6 +1,8 @@
1.2.6 1.2.6
* Write row markers when serializing schema (CASSANDRA-5572) * Write row markers when serializing schema (CASSANDRA-5572)
* Check only SSTables for the requested range when streaming (CASSANDRA-5569) * Check only SSTables for the requested range when streaming (CASSANDRA-5569)
Merged from 1.1
* Fix NPE in Pig's widerow mode (CASSANDRA-5488)
1.2.5 1.2.5

2
examples/pig/test/test_storage.pig
View File

@ -1,4 +1,4 @@
rows = LOAD 'cassandra://PigTest/SomeApp' USING CassandraStorage(); rows = LOAD 'cassandra://PigTest/SomeApp?widerows=true' USING CassandraStorage();
-- full copy -- full copy
STORE rows INTO 'cassandra://PigTest/CopyOfSomeApp' USING CassandraStorage(); STORE rows INTO 'cassandra://PigTest/CopyOfSomeApp' USING CassandraStorage();
-- single tuple -- single tuple

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

@ -147,7 +147,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
if (tuple.size() == 0) // lastRow is a new one if (tuple.size() == 0) // lastRow is a new one
{ {
key = (ByteBuffer)reader.getCurrentKey(); key = (ByteBuffer)reader.getCurrentKey();
addKeyToTuple(tuple, key, cfDef, parseType(cfDef.getKey_validation_class())); tuple = addKeyToTuple(tuple, key, cfDef, parseType(cfDef.getKey_validation_class()));
} }
for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet()) for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
{ {
@ -183,7 +183,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
key = (ByteBuffer)reader.getCurrentKey(); key = (ByteBuffer)reader.getCurrentKey();
if (lastKey != null && !(key.equals(lastKey))) // last key only had one value if (lastKey != null && !(key.equals(lastKey))) // last key only had one value
{ {
addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class())); tuple = addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet()) for (Map.Entry<ByteBuffer, IColumn> entry : lastRow.entrySet())
{ {
bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type()))); bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
@ -193,7 +193,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
lastRow = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue(); lastRow = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
return tuple; return tuple;
} }
addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class())); tuple = addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
} }
SortedMap<ByteBuffer,IColumn> row = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue(); SortedMap<ByteBuffer,IColumn> row = (SortedMap<ByteBuffer,IColumn>)reader.getCurrentValue();
if (lastRow != null) // prepend what was read last time if (lastRow != null) // prepend what was read last time
@ -236,7 +236,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
// output tuple, will hold the key, each indexed column in a tuple, then a bag of the rest // output tuple, will hold the key, each indexed column in a tuple, then a bag of the rest
// NOTE: we're setting the tuple size here only for the key so we can use setTupleValue on it // NOTE: we're setting the tuple size here only for the key so we can use setTupleValue on it
Tuple tuple = keyToTuple(key, cfDef, parseType(cfDef.getKey_validation_class())); Tuple tuple = addKeyToTuple(null, key, cfDef, parseType(cfDef.getKey_validation_class()));
DefaultDataBag bag = new DefaultDataBag(); DefaultDataBag bag = new DefaultDataBag();
// we must add all the indexed columns first to match the schema // we must add all the indexed columns first to match the schema
@ -295,15 +295,12 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
return t; return t;
} }
private Tuple keyToTuple(ByteBuffer key, CfDef cfDef, AbstractType comparator) throws IOException private Tuple addKeyToTuple(Tuple tuple, 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( tuple == null )
{
tuple = TupleFactory.getInstance().newTuple(1);
}
if( comparator instanceof AbstractCompositeType ) if( comparator instanceof AbstractCompositeType )
{ {
setTupleValue(tuple, 0, composeComposite((AbstractCompositeType)comparator,key)); setTupleValue(tuple, 0, composeComposite((AbstractCompositeType)comparator,key));
@ -312,7 +309,7 @@ public class CassandraStorage extends LoadFunc implements StoreFuncInterface, Lo
{ {
setTupleValue(tuple, 0, getDefaultMarshallers(cfDef).get(MarshallerType.KEY_VALIDATOR).compose(key)); setTupleValue(tuple, 0, getDefaultMarshallers(cfDef).get(MarshallerType.KEY_VALIDATOR).compose(key));
} }
return tuple;
} }
private Tuple columnToTuple(IColumn col, CfDef cfDef, AbstractType comparator) throws IOException private Tuple columnToTuple(IColumn col, CfDef cfDef, AbstractType comparator) throws IOException