cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [10/12] git commit: Pig support for hadoop CqlInputFormat
Date Wed, 23 Jul 2014 15:32:07 GMT
Pig support for hadoop CqlInputFormat

Patch by Alex Liu, reviewed by brandonwilliams for CASSANDRA-6454


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/62e70d23
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/62e70d23
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/62e70d23

Branch: refs/heads/cassandra-2.1
Commit: 62e70d23e5bacbcbdcf736fb60a02b543c965f99
Parents: a1fe634
Author: Brandon Williams <brandonwilliams@apache.org>
Authored: Wed Jul 23 10:28:46 2014 -0500
Committer: Brandon Williams <brandonwilliams@apache.org>
Committed: Wed Jul 23 10:28:46 2014 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../cassandra/hadoop/cql3/CqlConfigHelper.java  |   1 -
 .../cassandra/hadoop/cql3/CqlInputFormat.java   |   2 -
 .../cassandra/hadoop/pig/CqlNativeStorage.java  | 291 +++++++++++++++++++
 .../apache/cassandra/hadoop/pig/CqlStorage.java |  11 +-
 test/conf/cassandra.yaml                        |   2 +
 .../cassandra/pig/CqlTableDataTypeTest.java     |  94 +++++-
 .../org/apache/cassandra/pig/CqlTableTest.java  | 101 ++++++-
 .../org/apache/cassandra/pig/PigTestBase.java   |   3 +
 .../cassandra/pig/ThriftColumnFamilyTest.java   |  53 +++-
 10 files changed, 518 insertions(+), 41 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 028128d..6914a84 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.1
+ * Pig support for hadoop CqlInputFormat (CASSANDRA-6454)
  * Add listen_interface and rpc_interface options (CASSANDRA-7417)
  * Fail to start if commit log replay detects a problem (CASSANDRA-7125)
  * Improve schema merge performance (CASSANDRA-7444)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java
index 63279d1..b375ce2 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java
@@ -20,7 +20,6 @@ package org.apache.cassandra.hadoop.cql3;
 *
 */
 import java.io.FileInputStream;
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.security.KeyManagementException;
 import java.security.KeyStore;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
index e1cdf32..09bd80c 100644
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlInputFormat.java
@@ -18,8 +18,6 @@
 package org.apache.cassandra.hadoop.cql3;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Map;
 
 import org.apache.cassandra.hadoop.AbstractColumnFamilyInputFormat;
 import org.apache.hadoop.mapred.InputSplit;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
new file mode 100644
index 0000000..68249f9
--- /dev/null
+++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
@@ -0,0 +1,291 @@
+package org.apache.cassandra.hadoop.pig;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.cassandra.db.BufferCell;
+import org.apache.cassandra.db.Cell;
+import org.apache.cassandra.db.composites.CellNames;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.hadoop.ConfigHelper;
+import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
+import org.apache.cassandra.thrift.CfDef;
+import org.apache.cassandra.thrift.ColumnDef;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+
+import com.datastax.driver.core.Row;
+
+public class CqlNativeStorage extends CqlStorage
+{
+    private RecordReader<Long, Row> reader;
+    private String nativePort;
+    private String nativeCoreConnections;
+    private String nativeMaxConnections;
+    private String nativeMinSimultReqs;
+    private String nativeMaxSimultReqs;
+    private String nativeConnectionTimeout;
+    private String nativeReadConnectionTimeout;
+    private String nativeReceiveBufferSize;
+    private String nativeSendBufferSize;
+    private String nativeSolinger;
+    private String nativeTcpNodelay;
+    private String nativeReuseAddress;
+    private String nativeKeepAlive;
+    private String nativeAuthProvider;
+    private String nativeSSLTruststorePath;
+    private String nativeSSLKeystorePath;
+    private String nativeSSLTruststorePassword;
+    private String nativeSSLKeystorePassword;
+    private String nativeSSLCipherSuites;
+    private String inputCql;
+
+    public CqlNativeStorage()
+    {
+        this(1000);
+    }
+
+    /** @param pageSize limit number of CQL rows to fetch in a thrift request */
+    public CqlNativeStorage(int pageSize)
+    {
+        super(pageSize);
+        DEFAULT_INPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlInputFormat";
+    }
+
+    public void prepareToRead(RecordReader reader, PigSplit split)
+    {
+        this.reader = reader;
+    }
+
+    /** get next row */
+    public Tuple getNext() throws IOException
+    {
+        try
+        {
+            // load the next pair
+            if (!reader.nextKeyValue())
+                return null;
+
+            CfInfo cfInfo = getCfInfo(loadSignature);
+            CfDef cfDef = cfInfo.cfDef;
+            Row row = reader.getCurrentValue();
+            Tuple tuple = TupleFactory.getInstance().newTuple(cfDef.column_metadata.size());
+            Iterator<ColumnDef> itera = cfDef.column_metadata.iterator();
+            int i = 0;
+            while (itera.hasNext())
+            {
+                ColumnDef cdef = itera.next();
+                ByteBuffer columnValue = row.getBytesUnsafe(ByteBufferUtil.string(cdef.name.duplicate()));
+                if (columnValue != null)
+                {
+                    Cell cell = new BufferCell(CellNames.simpleDense(cdef.name), columnValue);
+                    AbstractType<?> validator = getValidatorMap(cfDef).get(cdef.name);
+                    setTupleValue(tuple, i, cqlColumnToObj(cell, cfDef), validator);
+                }
+                else
+                    tuple.set(i, null);
+                i++;
+            }
+            return tuple;
+        }
+        catch (InterruptedException e)
+        {
+            throw new IOException(e.getMessage());
+        }
+    }
+
+    /** set read configuration settings */
+    public void setLocation(String location, Job job) throws IOException
+    {
+        conf = job.getConfiguration();
+        setLocationFromUri(location);
+
+        if (username != null && password != null)
+        {
+            ConfigHelper.setInputKeyspaceUserNameAndPassword(conf, username, password);
+            CqlConfigHelper.setUserNameAndPassword(conf, username, password);
+        }
+        if (splitSize > 0)
+            ConfigHelper.setInputSplitSize(conf, splitSize);
+        if (partitionerClass!= null)
+            ConfigHelper.setInputPartitioner(conf, partitionerClass);
+        if (initHostAddress != null)
+            ConfigHelper.setInputInitialAddress(conf, initHostAddress);
+        if (rpcPort != null)
+            ConfigHelper.setInputRpcPort(conf, rpcPort);
+        if (nativePort != null)
+            CqlConfigHelper.setInputNativePort(conf, nativePort);
+        if (nativeCoreConnections != null)
+            CqlConfigHelper.setInputCoreConnections(conf, nativeCoreConnections);
+        if (nativeMaxConnections != null)
+            CqlConfigHelper.setInputMaxConnections(conf, nativeMaxConnections);
+        if (nativeMinSimultReqs != null)
+            CqlConfigHelper.setInputMinSimultReqPerConnections(conf, nativeMinSimultReqs);
+        if (nativeMinSimultReqs != null)
+            CqlConfigHelper.setInputMaxSimultReqPerConnections(conf, nativeMaxSimultReqs);
+        if (nativeConnectionTimeout != null)
+            CqlConfigHelper.setInputNativeConnectionTimeout(conf, nativeConnectionTimeout);
+        if (nativeReadConnectionTimeout != null)
+            CqlConfigHelper.setInputNativeReadConnectionTimeout(conf, nativeReadConnectionTimeout);
+        if (nativeReceiveBufferSize != null)
+            CqlConfigHelper.setInputNativeReceiveBufferSize(conf, nativeReceiveBufferSize);
+        if (nativeSendBufferSize != null)
+            CqlConfigHelper.setInputNativeSendBufferSize(conf, nativeSendBufferSize);
+        if (nativeSolinger != null)
+            CqlConfigHelper.setInputNativeSolinger(conf, nativeSolinger);
+        if (nativeTcpNodelay != null)
+            CqlConfigHelper.setInputNativeTcpNodelay(conf, nativeTcpNodelay);
+        if (nativeReuseAddress != null)
+            CqlConfigHelper.setInputNativeReuseAddress(conf, nativeReuseAddress);
+        if (nativeKeepAlive != null)
+            CqlConfigHelper.setInputNativeKeepAlive(conf, nativeKeepAlive);
+        if (nativeAuthProvider != null)
+            CqlConfigHelper.setInputNativeAuthProvider(conf, nativeAuthProvider);
+        if (nativeSSLTruststorePath != null)
+            CqlConfigHelper.setInputNativeSSLTruststorePath(conf, nativeSSLTruststorePath);
+        if (nativeSSLKeystorePath != null)
+            CqlConfigHelper.setInputNativeSSLKeystorePath(conf, nativeSSLKeystorePath);
+        if (nativeSSLTruststorePassword != null)
+            CqlConfigHelper.setInputNativeSSLTruststorePassword(conf, nativeSSLTruststorePassword);
+        if (nativeSSLKeystorePassword != null)
+            CqlConfigHelper.setInputNativeSSLKeystorePassword(conf, nativeSSLKeystorePassword);
+        if (nativeSSLCipherSuites != null)
+            CqlConfigHelper.setInputNativeSSLCipherSuites(conf, nativeSSLCipherSuites);
+
+        ConfigHelper.setInputColumnFamily(conf, keyspace, column_family);
+        setConnectionInformation();
+
+        CqlConfigHelper.setInputCQLPageRowSize(conf, String.valueOf(pageSize));
+        CqlConfigHelper.setInputCql(conf, inputCql);
+        if (System.getenv(PIG_INPUT_SPLIT_SIZE) != null)
+        {
+            try
+            {
+                ConfigHelper.setInputSplitSize(conf, Integer.parseInt(System.getenv(PIG_INPUT_SPLIT_SIZE)));
+            }
+            catch (NumberFormatException e)
+            {
+                throw new IOException("PIG_INPUT_SPLIT_SIZE is not a number", e);
+            }           
+        }
+
+        if (ConfigHelper.getInputInitialAddress(conf) == null)
+            throw new IOException("PIG_INPUT_INITIAL_ADDRESS or PIG_INITIAL_ADDRESS environment
variable not set");
+        if (ConfigHelper.getInputPartitioner(conf) == null)
+            throw new IOException("PIG_INPUT_PARTITIONER or PIG_PARTITIONER environment variable
not set");
+        if (loadSignature == null)
+            loadSignature = location;
+
+        initSchema(loadSignature);
+    }
+
+    private void setLocationFromUri(String location) throws IOException
+    {
+        try
+        {
+            if (!location.startsWith("cql://"))
+                throw new Exception("Bad scheme: " + location);
+
+            String[] urlParts = location.split("\\?");
+            if (urlParts.length > 1)
+            {
+                Map<String, String> urlQuery = getQueryMap(urlParts[1]);
+
+                // each page row size
+                if (urlQuery.containsKey("page_size"))
+                    pageSize = Integer.parseInt(urlQuery.get("page_size"));
+
+                // output prepared statement
+                if (urlQuery.containsKey("output_query"))
+                    outputQuery = urlQuery.get("output_query");
+
+                //split size
+                if (urlQuery.containsKey("split_size"))
+                    splitSize = Integer.parseInt(urlQuery.get("split_size"));
+                if (urlQuery.containsKey("partitioner"))
+                    partitionerClass = urlQuery.get("partitioner");
+                if (urlQuery.containsKey("use_secondary"))
+                    usePartitionFilter = Boolean.parseBoolean(urlQuery.get("use_secondary"));
+                if (urlQuery.containsKey("init_address"))
+                    initHostAddress = urlQuery.get("init_address");
+
+                if (urlQuery.containsKey("native_port"))
+                    nativePort = urlQuery.get("native_port");
+                if (urlQuery.containsKey("core_conns"))
+                    nativeCoreConnections = urlQuery.get("core_conns");
+                if (urlQuery.containsKey("max_conns"))
+                    nativeMaxConnections = urlQuery.get("max_conns");
+                if (urlQuery.containsKey("min_simult_reqs"))
+                    nativeMinSimultReqs = urlQuery.get("min_simult_reqs");
+                if (urlQuery.containsKey("max_simult_reqs"))
+                    nativeMaxSimultReqs = urlQuery.get("max_simult_reqs");
+                if (urlQuery.containsKey("native_timeout"))
+                    nativeConnectionTimeout = urlQuery.get("native_timeout");
+                if (urlQuery.containsKey("native_read_timeout"))
+                    nativeReadConnectionTimeout = urlQuery.get("native_read_timeout");
+                if (urlQuery.containsKey("rec_buff_size"))
+                    nativeReceiveBufferSize = urlQuery.get("rec_buff_size");
+                if (urlQuery.containsKey("send_buff_size"))
+                    nativeSendBufferSize = urlQuery.get("send_buff_size");
+                if (urlQuery.containsKey("solinger"))
+                    nativeSolinger = urlQuery.get("solinger");
+                if (urlQuery.containsKey("tcp_nodelay"))
+                    nativeTcpNodelay = urlQuery.get("tcp_nodelay");
+                if (urlQuery.containsKey("reuse_address"))
+                    nativeReuseAddress = urlQuery.get("reuse_address");
+                if (urlQuery.containsKey("keep_alive"))
+                    nativeKeepAlive = urlQuery.get("keep_alive");
+                if (urlQuery.containsKey("auth_provider"))
+                    nativeAuthProvider = urlQuery.get("auth_provider");
+                if (urlQuery.containsKey("trust_store_path"))
+                    nativeSSLTruststorePath = urlQuery.get("trust_store_path");
+                if (urlQuery.containsKey("key_store_path"))
+                    nativeSSLKeystorePath = urlQuery.get("key_store_path");
+                if (urlQuery.containsKey("trust_store_password"))
+                    nativeSSLTruststorePassword = urlQuery.get("trust_store_password");
+                if (urlQuery.containsKey("key_store_password"))
+                    nativeSSLKeystorePassword = urlQuery.get("key_store_password");
+                if (urlQuery.containsKey("cipher_suites"))
+                    nativeSSLCipherSuites = urlQuery.get("cipher_suites");
+                if (urlQuery.containsKey("input_cql"))
+                    inputCql = urlQuery.get("input_cql");
+                if (urlQuery.containsKey("rpc_port"))
+                    rpcPort = urlQuery.get("rpc_port");
+            }
+            String[] parts = urlParts[0].split("/+");
+            String[] credentialsAndKeyspace = parts[1].split("@");
+            if (credentialsAndKeyspace.length > 1)
+            {
+                String[] credentials = credentialsAndKeyspace[0].split(":");
+                username = credentials[0];
+                password = credentials[1];
+                keyspace = credentialsAndKeyspace[1];
+            }
+            else
+            {
+                keyspace = parts[1];
+            }
+            column_family = parts[2];
+        }
+        catch (Exception e)
+        {
+            throw new IOException("Expected 'cql://[username:password@]<keyspace>/<columnfamily>"
+
+                    "[?[page_size=<size>][&columns=<col1,col2>][&output_query=<prepared_statement>]"
+
+                    "[&where_clause=<clause>][&split_size=<size>][&partitioner=<partitioner>][&use_secondary=true|false]"
+
+                    "[&init_address=<host>][&native_port=<native_port>][&core_conns=<core_conns>]"
+
+                    "[&max_conns=<max_conns>][&min_simult_reqs=<min_simult_reqs>][&max_simult_reqs=<max_simult_reqs>]"
+
+                    "[&native_timeout=<native_timeout>][&native_read_timeout=<native_read_timeout>][&rec_buff_size=<rec_buff_size>]"
+
+                    "[&send_buff_size=<send_buff_size>][&solinger=<solinger>][&tcp_nodelay=<tcp_nodelay>][&reuse_address=<reuse_address>]"
+
+                    "[&keep_alive=<keep_alive>][&auth_provider=<auth_provider>][&trust_store_path=<trust_store_path>]"
+
+                    "[&key_store_path=<key_store_path>][&trust_store_password=<trust_store_password>]"
+
+                    "[&key_store_password=<key_store_password>][&cipher_suites=<cipher_suites>][&input_cql=<input_cql>]]':
" + e.getMessage());
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/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 6f17468..3e1d3d4 100644
--- a/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CqlStorage.java
@@ -34,7 +34,6 @@ import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
 import org.apache.cassandra.thrift.*;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
-
 import org.apache.hadoop.mapreduce.*;
 import org.apache.pig.Expression;
 import org.apache.pig.Expression.OpType;
@@ -58,11 +57,11 @@ public class CqlStorage extends AbstractCassandraStorage
 {
     private static final Logger logger = LoggerFactory.getLogger(CqlStorage.class);
     private RecordReader<Map<String, ByteBuffer>, Map<String, ByteBuffer>>
reader;
-    private RecordWriter<Map<String, ByteBuffer>, List<ByteBuffer>> writer;
+    protected RecordWriter<Map<String, ByteBuffer>, List<ByteBuffer>> writer;
 
-    private int pageSize = 1000;
+    protected int pageSize = 1000;
     private String columns;
-    private String outputQuery;
+    protected String outputQuery;
     private String whereClause;
     private boolean hasCompactValueAlias = false;
         
@@ -130,7 +129,7 @@ public class CqlStorage extends AbstractCassandraStorage
     }
 
     /** set the value to the position of the tuple */
-    private void setTupleValue(Tuple tuple, int position, Object value, AbstractType<?>
validator) throws ExecException
+    protected void setTupleValue(Tuple tuple, int position, Object value, AbstractType<?>
validator) throws ExecException
     {
         if (validator instanceof CollectionType)
             setCollectionTupleValues(tuple, position, value, validator);
@@ -184,7 +183,7 @@ public class CqlStorage extends AbstractCassandraStorage
     }
 
     /** convert a cql column to an object */
-    private Object cqlColumnToObj(Cell col, CfDef cfDef) throws IOException
+    protected Object cqlColumnToObj(Cell col, CfDef cfDef) throws IOException
     {
         // standard
         Map<ByteBuffer,AbstractType> validators = getValidatorMap(cfDef);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/test/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml
index 43995e6..1bc4193 100644
--- a/test/conf/cassandra.yaml
+++ b/test/conf/cassandra.yaml
@@ -35,3 +35,5 @@ server_encryption_options:
 incremental_backups: true
 concurrent_compactors: 4
 compaction_throughput_mb_per_sec: 0
+start_native_transport: true
+native_transport_port: 9052

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java b/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java
index 2020b0a..1819c61 100644
--- a/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java
+++ b/test/pig/org/apache/cassandra/pig/CqlTableDataTypeTest.java
@@ -220,7 +220,24 @@ public class CqlTableDataTypeTest extends PigTestBase
     public void testCqlStorageRegularType()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
-        pig.registerQuery("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + "'
USING CqlStorage();");
+        cqlTableTest("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + "' USING
CqlStorage();");
+        counterTableTest("cc_rows = LOAD 'cql://cql3ks/countertable?" + defaultParameters
+ "' USING CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageRegularType()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from cqltable where token(key) > ? and token(key) <= ?
+        cqlTableTest("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + nativeParameters
+ "&input_cql=select%20*%20from%20cqltable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
USING CqlStorage();");
+
+        //input_cql=select * from countertable where token(key) > ? and token(key) <=
?
+        counterTableTest("cc_rows = LOAD 'cql://cql3ks/countertable?" + defaultParameters
+ nativeParameters + "&input_cql=select%20*%20from%20countertable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
USING CqlStorage();");
+    }
+
+    private void cqlTableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
         Iterator<Tuple> it = pig.openIterator("rows");
         //{key: int, 
         //col_ascii: chararray, 
@@ -257,21 +274,45 @@ public class CqlTableDataTypeTest extends PigTestBase
             Assert.assertEquals(t.get(14), "varchar");
             Assert.assertEquals(t.get(15), 123);
         }
-        
-        pig.registerQuery("cc_rows = LOAD 'cql://cql3ks/countertable?" + defaultParameters
+ "' USING CqlStorage();");
-        it = pig.openIterator("cc_rows");
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
+        }
+    }
+
+    private void counterTableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
+        Iterator<Tuple>  it = pig.openIterator("cc_rows");
         if (it.hasNext()) {
             Tuple t = it.next();
             Assert.assertEquals(t.get(0), 1);
             Assert.assertEquals(t.get(1), 3L);
         }
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
+        }
     }
 
     @Test
     public void testCqlStorageSetType()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
-        pig.registerQuery("set_rows = LOAD 'cql://cql3ks/settable?" + defaultParameters +
"' USING CqlStorage();");
+        settableTest("set_rows = LOAD 'cql://cql3ks/settable?" + defaultParameters + "' USING
CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageSetType()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from settable where token(key) > ? and token(key) <= ?
+        settableTest("set_rows = LOAD 'cql://cql3ks/settable?" + defaultParameters + nativeParameters
+ "&input_cql=select%20*%20from%20settable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
USING CqlStorage();");
+    }
+
+    private void settableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
         Iterator<Tuple> it = pig.openIterator("set_rows");
         if (it.hasNext()) {
             Tuple t = it.next();
@@ -322,13 +363,30 @@ public class CqlTableDataTypeTest extends PigTestBase
             Assert.assertEquals(innerTuple.get(0), 123);
             Assert.assertEquals(innerTuple.get(1), 124);
         }
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
+        }
     }
 
     @Test
     public void testCqlStorageListType()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
-        pig.registerQuery("list_rows = LOAD 'cql://cql3ks/listtable?" + defaultParameters
+ "' USING CqlStorage();");
+        listtableTest("list_rows = LOAD 'cql://cql3ks/listtable?" + defaultParameters + "'
USING CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageListType()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from listtable where token(key) > ? and token(key) <=
?
+        listtableTest("list_rows = LOAD 'cql://cql3ks/listtable?" + defaultParameters + nativeParameters
+ "&input_cql=select%20*%20from%20listtable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
USING CqlStorage();");
+    }
+
+    private void listtableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
         Iterator<Tuple> it = pig.openIterator("list_rows");
         if (it.hasNext()) {
             Tuple t = it.next();
@@ -379,13 +437,30 @@ public class CqlTableDataTypeTest extends PigTestBase
             Assert.assertEquals(innerTuple.get(1), 123);
             Assert.assertEquals(innerTuple.get(0), 124);
         }
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
+        }
     }
 
     @Test
     public void testCqlStorageMapType()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
-        pig.registerQuery("map_rows = LOAD 'cql://cql3ks/maptable?" + defaultParameters +
"' USING CqlStorage();");
+        maptableTest("map_rows = LOAD 'cql://cql3ks/maptable?" + defaultParameters + "' USING
CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageMapType()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from maptable where token(key) > ? and token(key) <= ?
+        maptableTest("map_rows = LOAD 'cql://cql3ks/maptable?" + defaultParameters + nativeParameters
+ "&input_cql=select%20*%20from%20maptable%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
USING CqlStorage();");
+    }
+
+    private void maptableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
         Iterator<Tuple> it = pig.openIterator("map_rows");
         if (it.hasNext()) {
             Tuple t = it.next();
@@ -436,5 +511,10 @@ public class CqlTableDataTypeTest extends PigTestBase
             Assert.assertEquals(innerTuple.get(0), 123);
             Assert.assertEquals(innerTuple.get(1), 124);
         }
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
+        }
     }
+
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/test/pig/org/apache/cassandra/pig/CqlTableTest.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/CqlTableTest.java b/test/pig/org/apache/cassandra/pig/CqlTableTest.java
index 15d49f2..f5adef8 100644
--- a/test/pig/org/apache/cassandra/pig/CqlTableTest.java
+++ b/test/pig/org/apache/cassandra/pig/CqlTableTest.java
@@ -93,7 +93,41 @@ public class CqlTableTest extends PigTestBase
     public void testCqlStorageSchema()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
-        pig.registerQuery("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + "'
USING CqlStorage();");
+        cqlTableSchemaTest("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + "'
USING CqlStorage();");
+        compactCqlTableSchemaTest("rows = LOAD 'cql://cql3ks/compactcqltable?" + defaultParameters
+ "' USING CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageSchema()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from cqltable where token(key1) > ? and token(key1) <=
?
+        cqlTableSchemaTest("rows = LOAD 'cql://cql3ks/cqltable?" + defaultParameters + nativeParameters
+  "&input_cql=select%20*%20from%20cqltable%20where%20token(key1)%20%3E%20%3F%20and%20token(key1)%20%3C%3D%20%3F'
USING CqlNativeStorage();");
+
+        //input_cql=select * from compactcqltable where token(key1) > ? and token(key1)
<= ?
+        compactCqlTableSchemaTest("rows = LOAD 'cql://cql3ks/compactcqltable?" + defaultParameters
+ nativeParameters + "&input_cql=select%20*%20from%20compactcqltable%20where%20token(key1)%20%3E%20%3F%20and%20token(key1)%20%3C%3D%20%3F'
USING CqlNativeStorage();");
+    }
+
+    private void compactCqlTableSchemaTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
+        Iterator<Tuple>  it = pig.openIterator("rows");
+        if (it.hasNext()) {
+            Tuple t = it.next();
+            Assert.assertEquals(t.get(0).toString(), "key1");
+            Assert.assertEquals(t.get(1), 100);
+            Assert.assertEquals(t.get(2), 10.1f);
+            Assert.assertEquals(3, t.size());
+        }
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
+        }
+    }
+
+    private void cqlTableSchemaTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
         Iterator<Tuple> it = pig.openIterator("rows");
         if (it.hasNext()) {
             Tuple t = it.next();
@@ -103,15 +137,9 @@ public class CqlTableTest extends PigTestBase
             Assert.assertEquals(t.get(3), 10.1f);
             Assert.assertEquals(4, t.size());
         }
-
-        pig.registerQuery("rows = LOAD 'cql://cql3ks/compactcqltable?" + defaultParameters
+ "' USING CqlStorage();");
-        it = pig.openIterator("rows");
-        if (it.hasNext()) {
-            Tuple t = it.next();
-            Assert.assertEquals(t.get(0).toString(), "key1");
-            Assert.assertEquals(t.get(1), 100);
-            Assert.assertEquals(t.get(2), 10.1f);
-            Assert.assertEquals(3, t.size());
+        else
+        {
+            Assert.fail("Failed to get data for query " + initialQuery);
         }
     }
 
@@ -119,8 +147,23 @@ public class CqlTableTest extends PigTestBase
     public void testCqlStorageSingleKeyTable()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
+        SingleKeyTableTest("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters
+ "' USING CqlStorage();");
+
+    }
+
+    @Test
+    public void testCqlNativeStorageSingleKeyTable()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from moredata where token(x) > ? and token(x) <= ?
+        SingleKeyTableTest("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters
+ nativeParameters + "&input_cql=select%20*%20from%20moredata%20where%20token(x)%20%3E%20%3F%20and%20token(x)%20%3C%3D%20%3F'
USING CqlNativeStorage();");
+    }
+
+    private void SingleKeyTableTest(String initialQuery)
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
         pig.setBatchOn();
-        pig.registerQuery("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters
+ "' USING CqlStorage();");
+        pig.registerQuery(initialQuery);
         pig.registerQuery("insertformat= FOREACH moretestvalues GENERATE TOTUPLE(TOTUPLE('a',x)),TOTUPLE(y);");
         pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/test?" + defaultParameters
+ "&output_query=UPDATE+cql3ks.test+set+b+%3D+%3F' USING CqlStorage();");
         pig.executeBatch();
@@ -132,7 +175,7 @@ public class CqlTableTest extends PigTestBase
         //(1,1)
         pig.registerQuery("result= LOAD 'cql://cql3ks/test?" + defaultParameters + "' USING
CqlStorage();");
         Iterator<Tuple> it = pig.openIterator("result");
-        if (it.hasNext()) {
+        while (it.hasNext()) {
             Tuple t = it.next();
             Assert.assertEquals(t.get(0), t.get(1));
         }
@@ -142,8 +185,22 @@ public class CqlTableTest extends PigTestBase
     public void testCqlStorageCompositeKeyTable()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
+        CompositeKeyTableTest("moredata= LOAD 'cql://cql3ks/compmore?" + defaultParameters
+ "' USING CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageCompositeKeyTable()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from compmore where token(id) > ? and token(id) <= ?
+        CompositeKeyTableTest("moredata= LOAD 'cql://cql3ks/compmore?" + defaultParameters
+ nativeParameters + "&input_cql=select%20*%20from%20compmore%20where%20token(id)%20%3E%20%3F%20and%20token(id)%20%3C%3D%20%3F'
USING CqlNativeStorage();");
+    }
+
+    private void CompositeKeyTableTest(String initialQuery)
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
         pig.setBatchOn();
-        pig.registerQuery("moredata= LOAD 'cql://cql3ks/compmore?" + defaultParameters +
"' USING CqlStorage();");
+        pig.registerQuery(initialQuery);
         pig.registerQuery("insertformat = FOREACH moredata GENERATE TOTUPLE (TOTUPLE('a',x),TOTUPLE('b',y),
TOTUPLE('c',z)),TOTUPLE(data);");
         pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/compotable?" + defaultParameters
+ "&output_query=UPDATE%20cql3ks.compotable%20SET%20d%20%3D%20%3F' USING CqlStorage();");
         pig.executeBatch();
@@ -171,8 +228,22 @@ public class CqlTableTest extends PigTestBase
     public void testCqlStorageCollectionColumnTable()
     throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
     {
+        CollectionColumnTableTest("collectiontable= LOAD 'cql://cql3ks/collectiontable?"
+ defaultParameters + "' USING CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorageCollectionColumnTable()
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
+        //input_cql=select * from collectiontable where token(m) < ? and token(m) <=
?
+        CollectionColumnTableTest("collectiontable= LOAD 'cql://cql3ks/collectiontable?"
+ defaultParameters + nativeParameters + "&input_cql=input_cql%3Dselect%20*%20from%20collectiontable%20where%20token(m)%20%3C%20%3F%20and%20token(m)%20%3C%3D%20%3F'
USING CqlNativeStorage();");
+    }
+
+    private void CollectionColumnTableTest(String initialQuery)
+    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
+    {
         pig.setBatchOn();
-        pig.registerQuery("collectiontable= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters
+ "' USING CqlStorage();");
+        pig.registerQuery(initialQuery);
         pig.registerQuery("recs= FOREACH collectiontable GENERATE TOTUPLE(TOTUPLE('m', m)
), TOTUPLE(TOTUPLE('map', TOTUPLE('m', 'mm'), TOTUPLE('n', 'nn')));");
         pig.registerQuery("STORE recs INTO 'cql://cql3ks/collectiontable?" + defaultParameters
+ "&output_query=update+cql3ks.collectiontable+set+n+%3D+%3F' USING CqlStorage();");
         pig.executeBatch();
@@ -183,7 +254,7 @@ public class CqlTableTest extends PigTestBase
         //(book1,((m,mm),(n,nn)))
         pig.registerQuery("result= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters
+ "' USING CqlStorage();");
         Iterator<Tuple> it = pig.openIterator("result");
-        while (it.hasNext()) {
+        if (it.hasNext()) {
             Tuple t = it.next();
             Tuple t1 = (Tuple) t.get(1);
             Assert.assertEquals(t1.size(), 2);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/test/pig/org/apache/cassandra/pig/PigTestBase.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/PigTestBase.java b/test/pig/org/apache/cassandra/pig/PigTestBase.java
index 83dc63b..002fbba 100644
--- a/test/pig/org/apache/cassandra/pig/PigTestBase.java
+++ b/test/pig/org/apache/cassandra/pig/PigTestBase.java
@@ -66,6 +66,9 @@ public class PigTestBase extends SchemaLoader
     protected static MiniCluster cluster; 
     protected static PigServer pig;
     protected static String defaultParameters= "init_address=localhost&rpc_port=9170&partitioner=org.apache.cassandra.dht.ByteOrderedPartitioner";
+    protected static String nativeParameters = "&core_conns=2&max_conns=10&min_simult_reqs=3&max_simult_reqs=10&native_timeout=10000000"
 +
+                                               "&native_read_timeout=10000000&send_buff_size=4096&receive_buff_size=4096&solinger=3"
+
+                                               "&tcp_nodelay=true&reuse_address=true&keep_alive=true&native_port=9052";
 
     static
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/62e70d23/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
----------------------------------------------------------------------
diff --git a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
index 60344d2..b4a8662 100644
--- a/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
+++ b/test/pig/org/apache/cassandra/pig/ThriftColumnFamilyTest.java
@@ -203,7 +203,35 @@ public class ThriftColumnFamilyTest extends PigTestBase
     public void testCqlStorage() throws IOException, ClassNotFoundException, TException,
TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException,
IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException
     {
         //regular thrift column families
-        pig.registerQuery("data = load 'cql://thriftKs/SomeApp?" + defaultParameters + "'
using CqlStorage();");
+        cqlStorageTest("data = load 'cql://thriftKs/SomeApp?" + defaultParameters + "' using
CqlStorage();");
+
+        //Test counter colun family
+        // This test fails for CASSANDRA-7059
+        //cqlStorageCounterTableTest("cc_data = load 'cql://thriftKs/CC?" + defaultParameters
+ "' using CqlStorage();");
+
+        //Test composite column family
+        cqlStorageCompositeTableTest("compo_data = load 'cql://thriftKs/Compo?" + defaultParameters
+ "' using CqlStorage();");
+    }
+
+    @Test
+    public void testCqlNativeStorage() throws IOException, ClassNotFoundException, TException,
TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException,
IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException
+    {
+        //regular thrift column families
+        //input_cql=select * from "SomeApp" where token(key) > ? and token(key) <=
?
+        cqlStorageTest("data = load 'cql://thriftKs/SomeApp?" + defaultParameters + nativeParameters
+ "&input_cql=select%20*%20from%20%22SomeApp%22%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
using CqlNativeStorage();");
+
+        //Test counter colun family
+        //input_cql=select * from "CC" where token(key) > ? and token(key) <= ?
+        cqlStorageCounterTableTest("cc_data = load 'cql://thriftKs/CC?" + defaultParameters
+ nativeParameters + "&input_cql=select%20*%20from%20%22CC%22%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
using CqlNativeStorage();");
+
+        //Test composite column family
+        //input_cql=select * from "Compo" where token(key) > ? and token(key) <= ?
+        cqlStorageCompositeTableTest("compo_data = load 'cql://thriftKs/Compo?" + defaultParameters
+ nativeParameters + "&input_cql=select%20*%20from%20%22Compo%22%20where%20token(key)%20%3E%20%3F%20and%20token(key)%20%3C%3D%20%3F'
using CqlNativeStorage();");
+    }
+
+    private void cqlStorageTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
 
         //(bar,3.141592653589793,1335890877,User Bar,35.0,9,15000,like)
         //(baz,1.61803399,1335890877,User Baz,95.3,3,512000,dislike)
@@ -256,16 +284,18 @@ public class ThriftColumnFamilyTest extends PigTestBase
             }
         }
         Assert.assertEquals(count, 4);
+    }
 
-        //Test counter colun family
-        pig.registerQuery("cc_data = load 'cql://thriftKs/CC?" + defaultParameters + "' using
CqlStorage();");
+    private void cqlStorageCounterTableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
 
         //(chuck,fist,1)
         //(chuck,kick,3)
 
         // {key: chararray,column1: chararray,value: long}
-        it = pig.openIterator("cc_data");
-        count = 0;
+        Iterator<Tuple> it = pig.openIterator("cc_data");
+        int count = 0;
         while (it.hasNext()) {
             count ++;
             Tuple t = it.next();
@@ -275,9 +305,11 @@ public class ThriftColumnFamilyTest extends PigTestBase
                 Assert.assertEquals(t.get(2), 3L);
         }
         Assert.assertEquals(count, 2);
+    }
 
-        //Test composite column family
-        pig.registerQuery("compo_data = load 'cql://thriftKs/Compo?" + defaultParameters
+ "' using CqlStorage();");
+    private void cqlStorageCompositeTableTest(String initialQuery) throws IOException
+    {
+        pig.registerQuery(initialQuery);
 
         //(kick,bruce,bruce,watch it, mate)
         //(kick,bruce,lee,oww)
@@ -285,8 +317,8 @@ public class ThriftColumnFamilyTest extends PigTestBase
         //(punch,bruce,lee,ouch)
 
         //{key: chararray,column1: chararray,column2: chararray,value: chararray}
-        it = pig.openIterator("compo_data");
-        count = 0;
+        Iterator<Tuple> it = pig.openIterator("compo_data");
+        int count = 0;
         while (it.hasNext()) {
             count ++;
             Tuple t = it.next();
@@ -583,7 +615,8 @@ public class ThriftColumnFamilyTest extends PigTestBase
         }
     }
 
-    @Test
+    /** This test case fails due to antlr lib conflicts, Cassandra2.1 uses 3.2, Hive1.2 uses
3.4 */
+    //@Test
     public void testCassandraStorageCompositeColumnCF() throws IOException, ClassNotFoundException,
TException, TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException,
UnavailableException, IllegalAccessException, InstantiationException, AuthenticationException,
AuthorizationException
     {
         //Test CompositeType


Mime
View raw message