Merge branch 'cassandra-2.1' into trunk

src/java/org/apache/cassandra/hadoop/cql3/CqlConfigHelper.java

@@ -81,6 +81,8 @@ public class CqlConfigHelper
     private static final String INPUT_NATIVE_SSL_KEY_STORE_PASSWARD = "cassandra.input.native.ssl.key.store.password";
     private static final String INPUT_NATIVE_SSL_CIPHER_SUITES = "cassandra.input.native.ssl.cipher.suites";
 
+    private static final String INPUT_NATIVE_PROTOCOL_VERSION = "cassandra.input.native.protocol.version";
+
     private static final String OUTPUT_CQL = "cassandra.output.cql";
     
     /**
@@ -279,6 +281,10 @@ public class CqlConfigHelper
         return conf.get(OUTPUT_CQL);
     }
 
+    private static Optional<Integer> getProtocolVersion(Configuration conf) {
+        return getIntSetting(INPUT_NATIVE_PROTOCOL_VERSION, conf);
+    }
+
     public static Cluster getInputCluster(String host, Configuration conf)
     {
         // this method has been left for backward compatibility
@@ -290,6 +296,7 @@ public class CqlConfigHelper
         int port = getInputNativePort(conf);
         Optional<AuthProvider> authProvider = getAuthProvider(conf);
         Optional<SSLOptions> sslOptions = getSSLOptions(conf);
+        Optional<Integer> protocolVersion = getProtocolVersion(conf);
         LoadBalancingPolicy loadBalancingPolicy = getReadLoadBalancingPolicy(conf, hosts);
         SocketOptions socketOptions = getReadSocketOptions(conf);
         QueryOptions queryOptions = getReadQueryOptions(conf);
@@ -305,6 +312,9 @@ public class CqlConfigHelper
         if (sslOptions.isPresent())
             builder.withSSL(sslOptions.get());
 
+        if (protocolVersion.isPresent()) {
+            builder.withProtocolVersion(protocolVersion.get());
+        }
         builder.withLoadBalancingPolicy(loadBalancingPolicy)
                .withSocketOptions(socketOptions)
                .withQueryOptions(queryOptions)
@@ -313,6 +323,7 @@ public class CqlConfigHelper
         return builder.build();
     }
 
+
     public static void setInputCoreConnections(Configuration conf, String connections)
     {
         conf.set(INPUT_NATIVE_CORE_CONNECTIONS_PER_HOST, connections);

src/java/org/apache/cassandra/hadoop/cql3/CqlRecordReader.java

@@ -92,6 +92,7 @@ public class CqlRecordReader extends RecordReader<Long, Row>
 
     // partition keys -- key aliases
     private LinkedHashMap<String, Boolean> partitionBoundColumns = Maps.newLinkedHashMap();
+    protected int nativeProtocolVersion = 1;
 
     public CqlRecordReader()
     {
@@ -132,6 +133,9 @@ public class CqlRecordReader extends RecordReader<Long, Row>
         if (session == null)
           throw new RuntimeException("Can't create connection session");
 
+        //get negotiated serialization protocol
+        nativeProtocolVersion = cluster.getConfiguration().getProtocolOptions().getProtocolVersion();
+
         // If the user provides a CQL query then we will use it without validation
         // otherwise we will fall back to building a query using the:
         //   inputColumns
@@ -233,6 +237,14 @@ public class CqlRecordReader extends RecordReader<Long, Row>
         return new WrappedRow();
     }
 
+    /**
+     * Return native version protocol of the cluster connection
+     * @return serialization protocol version.
+     */
+    public int getNativeProtocolVersion() {
+        return nativeProtocolVersion;
+    }
+
     /** CQL row iterator 
      *  Input cql query  
      *  1) select clause must include key columns (if we use partition key based row count)

src/java/org/apache/cassandra/hadoop/pig/AbstractCassandraStorage.java

@@ -62,6 +62,7 @@ import org.slf4j.LoggerFactory;
  */
 public abstract class AbstractCassandraStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata
 {
+
     protected enum MarshallerType { COMPARATOR, DEFAULT_VALIDATOR, KEY_VALIDATOR, SUBCOMPARATOR };
 
     // system environment variables that can be set to configure connection info:
@@ -101,6 +102,8 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
     protected boolean usePartitionFilter = false;
     protected String initHostAddress;
     protected String rpcPort;
+    protected int nativeProtocolVersion = 1;
+
 
     public AbstractCassandraStorage()
     {
@@ -785,7 +788,7 @@ public abstract class AbstractCassandraStorage extends LoadFunc implements Store
         {
             // For CollectionType, the compose() method assumes the v3 protocol format of collection, which
             // is not correct here since we query using the CQL-over-thrift interface which use the pre-v3 format
-            return ((CollectionSerializer)validator.getSerializer()).deserializeForNativeProtocol(value, 1);
+            return ((CollectionSerializer)validator.getSerializer()).deserializeForNativeProtocol(value, nativeProtocolVersion);
         }
 
         return validator.compose(value);

src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java

@@ -28,6 +28,7 @@ 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.hadoop.cql3.CqlRecordReader;
 import org.apache.cassandra.thrift.CfDef;
 import org.apache.cassandra.thrift.ColumnDef;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -78,6 +79,9 @@ public class CqlNativeStorage extends CqlStorage
     public void prepareToRead(RecordReader reader, PigSplit split)
     {
         this.reader = reader;
+        if (reader instanceof CqlRecordReader) {
+            nativeProtocolVersion = ((CqlRecordReader) reader).getNativeProtocolVersion();
+        }
     }
 
     /** get next row */