You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by br...@apache.org on 2013/10/12 00:43:25 UTC
[3/5] git commit: Revert "Merge branch 'cassandra-1.2' into
cassandra-2.0"
Revert "Merge branch 'cassandra-1.2' into cassandra-2.0"
This reverts commit a3ad2e82249b88d4a05f24140948cdbc809d14f3, reversing
changes made to 8a506e66a66c004a7a253e3dd28845517da8a967.
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e5dba3c6
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e5dba3c6
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e5dba3c6
Branch: refs/heads/trunk
Commit: e5dba3c6250eb7285777490a937b66a7092afa77
Parents: a3ad2e8
Author: Brandon Williams <br...@apache.org>
Authored: Fri Oct 11 17:41:22 2013 -0500
Committer: Brandon Williams <br...@apache.org>
Committed: Fri Oct 11 17:41:22 2013 -0500
----------------------------------------------------------------------
.../hadoop/pig/AbstractCassandraStorage.java | 97 ++++++--------------
.../cassandra/hadoop/pig/CassandraStorage.java | 55 ++++-------
.../apache/cassandra/hadoop/pig/CqlStorage.java | 8 +-
3 files changed, 51 insertions(+), 109 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e5dba3c6/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
index 486c781..c881734 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java
@@ -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,9 +116,8 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
}
/** convert a column to a tuple */
- protected Tuple columnToTuple(IColumn col, CfInfo cfInfo, AbstractType comparator) throws IOException
+ protected Tuple columnToTuple(Column col, CfDef cfDef, AbstractType comparator) throws IOException
{
- CfDef cfDef = cfInfo.cfDef;
Tuple pair = TupleFactory.getInstance().newTuple(2);
// name
@@ -131,34 +130,11 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
if (validators.get(col.name()) == null)
{
- // standard
- Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
- 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(colName), col.value()));
- return pair;
+ Map<MarshallerType, AbstractType> marshallers = getDefaultMarshallers(cfDef);
+ setTupleValue(pair, 1, cassandraToObj(marshallers.get(MarshallerType.DEFAULT_VALIDATOR), col.value()));
}
else
- {
- // super
- ArrayList<Tuple> subcols = new ArrayList<Tuple>();
- for (IColumn subcol : col.getSubColumns())
- subcols.add(columnToTuple(subcol, cfInfo, parseType(cfDef.getSubcomparator_type())));
-
- pair.set(1, new DefaultDataBag(subcols));
- }
+ setTupleValue(pair, 1, cassandraToObj(validators.get(col.name()), col.value()));
return pair;
}
@@ -178,16 +154,11 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
}
/** get the columnfamily definition for the signature */
- protected CfInfo getCfInfo(String signature) throws IOException
+ protected CfDef getCfDef(String signature) throws IOException
{
UDFContext context = UDFContext.getUDFContext();
Properties property = context.getUDFProperties(AbstractCassandraStorage.class);
- 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;
+ return cfdefFromString(property.getProperty(signature));
}
/** construct a map to store the mashaller type to cassandra data type mapping */
@@ -344,7 +315,10 @@ 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 DecimalType ||
+ type instanceof InetAddressType ||
+ type instanceof LexicalUUIDType ||
+ type instanceof UUIDType )
return DataType.CHARARRAY;
else if (type instanceof FloatType)
return DataType.FLOAT;
@@ -525,15 +499,11 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
}
}
- // compose the CfInfo for the columfamily
- CfInfo cfInfo = getCfInfo(client);
+ // compose the CfDef for the columfamily
+ CfDef cfDef = getCfDef(client);
- 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());
- }
+ if (cfDef != null)
+ properties.setProperty(signature, cfdefToString(cfDef));
else
throw new IOException(String.format("Column family '%s' not found in keyspace '%s'",
column_family,
@@ -579,17 +549,17 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
return cfDef;
}
- /** return the CfInf for the column family */
- protected CfInfo getCfInfo(Cassandra.Client client)
+ /** return the CfDef for the column family */
+ protected CfDef getCfDef(Cassandra.Client client)
throws InvalidRequestException,
UnavailableException,
TimedOutException,
SchemaDisagreementException,
TException,
+ CharacterCodingException,
NotFoundException,
org.apache.cassandra.exceptions.InvalidRequestException,
- ConfigurationException,
- IOException
+ ConfigurationException
{
// get CF meta data
String query = "SELECT type," +
@@ -643,19 +613,12 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
else
cql3Table = true;
}
- 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;
+ cfDef.column_metadata = getColumnMetadata(client, cql3Table);
+ return cfDef;
}
/** get a list of columns */
- protected abstract List<ColumnDef> getColumnMetadata(Cassandra.Client client)
+ protected abstract List<ColumnDef> getColumnMetadata(Cassandra.Client client, boolean cql3Table)
throws InvalidRequestException,
UnavailableException,
TimedOutException,
@@ -772,7 +735,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 = getCfInfo(loadSignature).cfDef;
+ CfDef cfdef = getCfDef(loadSignature);
List<ColumnDef> indexes = new ArrayList<ColumnDef>();
for (ColumnDef cdef : cfdef.column_metadata)
{
@@ -801,17 +764,15 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
protected Object cassandraToObj(AbstractType validator, ByteBuffer value)
{
- if (validator instanceof DecimalType || validator instanceof InetAddressType)
+ if (validator instanceof DecimalType ||
+ validator instanceof InetAddressType ||
+ validator instanceof LexicalUUIDType ||
+ validator instanceof UUIDType)
+ {
return validator.getString(value);
+ }
else
return validator.compose(value);
}
-
- protected class CfInfo
- {
- boolean compactCqlTable = false;
- boolean cql3Table = false;
- CfDef cfDef;
- }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e5dba3c6/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
index d9c55a1..4083236 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
@@ -103,8 +103,7 @@ public class CassandraStorage extends AbstractCassandraStorage
/** read wide row*/
public Tuple getNextWide() throws IOException
{
- CfInfo cfInfo = getCfInfo(loadSignature);
- CfDef cfDef = cfInfo.cfDef;
+ CfDef cfDef = getCfDef(loadSignature);
ByteBuffer key = null;
Tuple tuple = null;
DefaultDataBag bag = new DefaultDataBag();
@@ -127,7 +126,7 @@ public class CassandraStorage extends AbstractCassandraStorage
}
for (Map.Entry<ByteBuffer, Column> entry : lastRow.entrySet())
{
- bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
+ bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
}
lastKey = null;
lastRow = null;
@@ -165,7 +164,7 @@ public class CassandraStorage extends AbstractCassandraStorage
addKeyToTuple(tuple, lastKey, cfDef, parseType(cfDef.getKey_validation_class()));
for (Map.Entry<ByteBuffer, Column> entry : lastRow.entrySet())
{
- bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
+ bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
}
tuple.append(bag);
lastKey = key;
@@ -182,14 +181,14 @@ public class CassandraStorage extends AbstractCassandraStorage
{
for (Map.Entry<ByteBuffer, Column> entry : lastRow.entrySet())
{
- bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
+ bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
}
lastKey = null;
lastRow = null;
}
for (Map.Entry<ByteBuffer, Column> entry : row.entrySet())
{
- bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
+ bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
}
}
}
@@ -211,8 +210,7 @@ public class CassandraStorage extends AbstractCassandraStorage
if (!reader.nextKeyValue())
return null;
- CfInfo cfInfo = getCfInfo(loadSignature);
- CfDef cfDef = cfInfo.cfDef;
+ CfDef cfDef = getCfDef(loadSignature);
ByteBuffer key = reader.getCurrentKey();
Map<ByteBuffer, Column> cf = reader.getCurrentValue();
assert key != null && cf != null;
@@ -228,21 +226,11 @@ public class CassandraStorage extends AbstractCassandraStorage
// take care to iterate these in the same order as the schema does
for (ColumnDef cdef : cfDef.column_metadata)
{
- boolean hasColumn = false;
- boolean cql3Table = false;
- try
+ if (cf.containsKey(cdef.name))
{
- hasColumn = cf.containsKey(cdef.name);
+ tuple.append(columnToTuple(cf.get(cdef.name), cfDef, parseType(cfDef.getComparator_type())));
}
- catch (Exception e)
- {
- cql3Table = true;
- }
- if (hasColumn)
- {
- tuple.append(columnToTuple(cf.get(cdef.name), cfInfo, parseType(cfDef.getComparator_type())));
- }
- else if (!cql3Table)
+ else
{ // otherwise, we need to add an empty tuple to take its place
tuple.append(TupleFactory.getInstance().newTuple());
}
@@ -252,7 +240,7 @@ public class CassandraStorage extends AbstractCassandraStorage
for (Map.Entry<ByteBuffer, Column> entry : cf.entrySet())
{
if (!added.containsKey(entry.getKey()))
- bag.add(columnToTuple(entry.getValue(), cfInfo, parseType(cfDef.getComparator_type())));
+ bag.add(columnToTuple(entry.getValue(), cfDef, parseType(cfDef.getComparator_type())));
}
tuple.append(bag);
// finally, special top-level indexes if needed
@@ -260,7 +248,7 @@ public class CassandraStorage extends AbstractCassandraStorage
{
for (ColumnDef cdef : getIndexes())
{
- Tuple throwaway = columnToTuple(cf.get(cdef.name), cfInfo, parseType(cfDef.getComparator_type()));
+ Tuple throwaway = columnToTuple(cf.get(cdef.name), cfDef, parseType(cfDef.getComparator_type()));
tuple.append(throwaway.get(1));
}
}
@@ -370,8 +358,7 @@ public class CassandraStorage extends AbstractCassandraStorage
public ResourceSchema getSchema(String location, Job job) throws IOException
{
setLocation(location, job);
- CfInfo cfInfo = getCfInfo(loadSignature);
- CfDef cfDef = cfInfo.cfDef;
+ CfDef cfDef = getCfDef(loadSignature);
if (cfDef.column_type.equals("Super"))
return null;
@@ -418,7 +405,7 @@ public class CassandraStorage extends AbstractCassandraStorage
// add the key first, then the indexed columns, and finally the bag
allSchemaFields.add(keyFieldSchema);
- if (!widerows && (cfInfo.compactCqlTable || !cfInfo.cql3Table))
+ if (!widerows)
{
// defined validators/indexes
for (ColumnDef cdef : cfDef.column_metadata)
@@ -712,17 +699,12 @@ public class CassandraStorage extends AbstractCassandraStorage
}
/** get a list of column for the column family */
- protected List<ColumnDef> getColumnMetadata(Cassandra.Client client)
- throws InvalidRequestException,
- UnavailableException,
- TimedOutException,
- SchemaDisagreementException,
- TException,
- CharacterCodingException,
- org.apache.cassandra.exceptions.InvalidRequestException,
- ConfigurationException,
- NotFoundException
+ protected List<ColumnDef> getColumnMetadata(Cassandra.Client client, boolean cql3Table)
+ throws TException, CharacterCodingException, InvalidRequestException, ConfigurationException
{
+ if (cql3Table)
+ return new ArrayList<>();
+
return getColumnMeta(client, true, true);
}
@@ -807,5 +789,6 @@ public class CassandraStorage extends AbstractCassandraStorage
"[&comparator=<comparator>][&split_size=<size>][&partitioner=<partitioner>]]': " + e.getMessage());
}
}
+
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/e5dba3c6/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
index 9a9b522..50ee6b7 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
@@ -94,8 +94,7 @@ public class CqlStorage extends AbstractCassandraStorage
if (!reader.nextKeyValue())
return null;
- CfInfo cfInfo = getCfInfo(loadSignature);
- CfDef cfDef = cfInfo.cfDef;
+ CfDef cfDef = getCfDef(loadSignature);
Map<String, ByteBuffer> keys = reader.getCurrentKey();
Map<String, ByteBuffer> columns = reader.getCurrentValue();
assert keys != null && columns != null;
@@ -281,8 +280,7 @@ public class CqlStorage extends AbstractCassandraStorage
public ResourceSchema getSchema(String location, Job job) throws IOException
{
setLocation(location, job);
- CfInfo cfInfo = getCfInfo(loadSignature);
- CfDef cfDef = cfInfo.cfDef;
+ CfDef cfDef = getCfDef(loadSignature);
// top-level schema, no type
ResourceSchema schema = new ResourceSchema();
@@ -432,7 +430,7 @@ public class CqlStorage extends AbstractCassandraStorage
}
/** include key columns */
- protected List<ColumnDef> getColumnMetadata(Cassandra.Client client)
+ protected List<ColumnDef> getColumnMetadata(Cassandra.Client client, boolean cql3Table)
throws InvalidRequestException,
UnavailableException,
TimedOutException,