cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [3/5] git commit: Revert "Merge branch 'cassandra-1.2' into cassandra-2.0"
Date Fri, 11 Oct 2013 22:43:25 GMT
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 <brandonwilliams@apache.org>
Authored: Fri Oct 11 17:41:22 2013 -0500
Committer: Brandon Williams <brandonwilliams@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,


Mime
View raw message