cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject svn commit: r1177083 - /cassandra/branches/cassandra-0.8/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
Date Wed, 28 Sep 2011 22:01:02 GMT
Author: brandonwilliams
Date: Wed Sep 28 22:01:01 2011
New Revision: 1177083

URL: http://svn.apache.org/viewvc?rev=1177083&view=rev
Log:
Pig storage handler implements LoadMetadata interface.
Patch by brandonwilliams, reviewed by Jeremy Hanna for CASSANDRA-2777

Modified:
    cassandra/branches/cassandra-0.8/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java

Modified: cassandra/branches/cassandra-0.8/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java?rev=1177083&r1=1177082&r2=1177083&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
(original)
+++ cassandra/branches/cassandra-0.8/contrib/pig/src/java/org/apache/cassandra/hadoop/pig/CassandraStorage.java
Wed Sep 28 22:01:01 2011
@@ -31,7 +31,7 @@ import org.apache.commons.logging.LogFac
 import org.apache.cassandra.db.Column;
 import org.apache.cassandra.db.IColumn;
 import org.apache.cassandra.db.SuperColumn;
-import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.hadoop.*;
 import org.apache.cassandra.thrift.Mutation;
 import org.apache.cassandra.thrift.Deletion;
@@ -46,6 +46,7 @@ import org.apache.pig.*;
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
 import org.apache.pig.data.*;
+import org.apache.pig.ResourceSchema.ResourceFieldSchema;
 import org.apache.pig.impl.logicalLayer.FrontendException;
 import org.apache.pig.impl.util.UDFContext;
 import org.apache.thrift.TDeserializer;
@@ -61,7 +62,7 @@ import org.apache.thrift.transport.TTran
  *
  * A row from a standard CF will be returned as nested tuples: (key, ((name1, val1), (name2,
val2))).
  */
-public class CassandraStorage extends LoadFunc implements StoreFuncInterface
+public class CassandraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata
 {
     // system environment variables that can be set to configure connection info:
     // alternatively, Hadoop JobConf variables can be set using keys from ConfigHelper
@@ -293,6 +294,103 @@ public class CassandraStorage extends Lo
         initSchema(loadSignature);
     }
 
+    public ResourceSchema getSchema(String location, Job job) throws IOException
+    {
+        setLocation(location, job);
+        CfDef cfDef = getCfDef(loadSignature);
+
+        if (cfDef.column_type.equals("Super"))
+            return null;
+        // top-level schema, no type
+        ResourceSchema schema = new ResourceSchema();
+
+        // add key
+        ResourceFieldSchema keyFieldSchema = new ResourceFieldSchema();
+        keyFieldSchema.setName("key");
+        keyFieldSchema.setType(DataType.CHARARRAY); //TODO: get key type
+
+        // will become the bag of tuples
+        ResourceFieldSchema bagFieldSchema = new ResourceFieldSchema();
+        bagFieldSchema.setName("columns");
+        bagFieldSchema.setType(DataType.BAG);
+        ResourceSchema bagSchema = new ResourceSchema();
+
+
+        List<AbstractType> marshallers = getDefaultMarshallers(cfDef);
+        Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);
+        List<ResourceFieldSchema> tupleFields = new ArrayList<ResourceFieldSchema>();
+
+        // default comparator/validator
+        ResourceSchema innerTupleSchema = new ResourceSchema();
+        ResourceFieldSchema tupleField = new ResourceFieldSchema();
+        tupleField.setType(DataType.TUPLE);
+        tupleField.setSchema(innerTupleSchema);
+
+        ResourceFieldSchema colSchema = new ResourceFieldSchema();
+        colSchema.setName("name");
+        colSchema.setType(getPigType(marshallers.get(0)));
+        tupleFields.add(colSchema);
+
+        ResourceFieldSchema valSchema = new ResourceFieldSchema();
+        AbstractType validator = marshallers.get(1);
+        valSchema.setName("value");
+        valSchema.setType(getPigType(validator));
+        tupleFields.add(valSchema);
+
+        // defined validators/indexes
+        for (ColumnDef cdef : cfDef.column_metadata)
+        {
+            colSchema = new ResourceFieldSchema();
+            colSchema.setName(new String(cdef.getName()));
+            colSchema.setType(getPigType(marshallers.get(0)));
+            tupleFields.add(colSchema);
+
+            valSchema = new ResourceFieldSchema();
+            validator = validators.get(cdef.getName());
+            if (validator == null)
+                validator = marshallers.get(1);
+            valSchema.setName("value");
+            valSchema.setType(getPigType(validator));
+            tupleFields.add(valSchema);
+        }
+        innerTupleSchema.setFields(tupleFields.toArray(new ResourceFieldSchema[tupleFields.size()]));
+
+        // a bag can contain only one tuple, but that tuple can contain anything
+        bagSchema.setFields(new ResourceFieldSchema[] { tupleField });
+        bagFieldSchema.setSchema(bagSchema);
+        // top level schema contains everything
+        schema.setFields(new ResourceFieldSchema[] { keyFieldSchema, bagFieldSchema });
+        return schema;
+    }
+
+    private byte getPigType(AbstractType type)
+    {
+        if (type instanceof LongType)
+            return DataType.LONG;
+        else if (type instanceof IntegerType)
+            return DataType.INTEGER;
+        else if (type instanceof AsciiType)
+            return DataType.CHARARRAY;
+        else if (type instanceof UTF8Type)
+            return DataType.CHARARRAY;
+        return DataType.BYTEARRAY;
+    }
+
+    public ResourceStatistics getStatistics(String location, Job job)
+    {
+        return null;
+    }
+
+    public String[] getPartitionKeys(String location, Job job)
+    {
+        return null;
+    }
+
+    public void setPartitionFilter(Expression partitionFilter)
+    {
+        // no-op
+    }
+
     @Override
     public String relativeToAbsolutePath(String location, Path curDir) throws IOException
     {



Mime
View raw message