From 639c01a3504ba2e2a55061093651a9973ad68d11 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Fri, 11 Oct 2013 15:22:35 -0500 Subject: [PATCH] Fix int/bigint in CassandraStorage Patch by Alex Liu, reviewed by brandonwilliams for CASSANDRA-6102 --- .../hadoop/pig/AbstractCassandraStorage.java | 82 ++++++++++++------- .../hadoop/pig/CassandraStorage.java | 45 ++++++---- .../cassandra/hadoop/pig/CqlStorage.java | 8 +- 3 files changed, 85 insertions(+), 50 deletions(-) diff --git a/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java index 6ad4f9ef3e..dbebfb561a 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,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 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 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 subcols = new ArrayList(); 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 getColumnMetadata(Cassandra.Client client, boolean cql3Table) + protected abstract List 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 getIndexes() throws IOException { - CfDef cfdef = getCfDef(loadSignature); + CfDef cfdef = getCfInfo(loadSignature).cfDef; List indexes = new ArrayList(); 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; + } } diff --git a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java index 1135b703fd..a7cc1ad900 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java @@ -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 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 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 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 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 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 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 getColumnMetadata(Cassandra.Client client, boolean cql3Table) + protected List getColumnMetadata(Cassandra.Client client) throws InvalidRequestException, UnavailableException, TimedOutException, @@ -708,9 +721,6 @@ public class CassandraStorage extends AbstractCassandraStorage ConfigurationException, NotFoundException { - if (cql3Table) - return new ArrayList(); - return getColumnMeta(client, true, true); } @@ -795,6 +805,5 @@ public class CassandraStorage extends AbstractCassandraStorage "[&comparator=][&split_size=][&partitioner=]]': " + e.getMessage()); } } - } diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java index e51338c509..7e1f56c7fe 100644 --- a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java +++ b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java @@ -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 keys = reader.getCurrentKey(); Map 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 getColumnMetadata(Cassandra.Client client, boolean cql3Table) + protected List getColumnMetadata(Cassandra.Client client) throws InvalidRequestException, UnavailableException, TimedOutException,