cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alek...@apache.org
Subject [2/4] cassandra git commit: Merge branch 'cassandra-2.1' into cassandra-2.2
Date Tue, 15 Sep 2015 15:17:06 GMT
Merge branch 'cassandra-2.1' into cassandra-2.2


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

Branch: refs/heads/trunk
Commit: cd4a1e6acc51b0f708127a50d37dd7832bbe8dfa
Parents: 9791796 c7b4073
Author: Aleksey Yeschenko <aleksey@apache.org>
Authored: Tue Sep 15 16:12:45 2015 +0100
Committer: Aleksey Yeschenko <aleksey@apache.org>
Committed: Tue Sep 15 16:12:45 2015 +0100

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../hadoop/AbstractColumnFamilyInputFormat.java |   2 +
 .../hadoop/cql3/CqlBulkRecordWriter.java        |  13 +-
 .../cassandra/hadoop/pig/CqlNativeStorage.java  | 171 ++++++++++++-------
 test/conf/cassandra_pig.yaml                    |  41 +++++
 .../org/apache/cassandra/pig/CqlTableTest.java  |  35 ++++
 .../org/apache/cassandra/pig/PigTestBase.java   |   3 +-
 7 files changed, 205 insertions(+), 61 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index bd24781,5f11049..b0ade42
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,17 -1,8 +1,18 @@@
 -2.1.10
 +2.2.2
 + * Defer default role manager setup until all nodes are on 2.2+ (CASSANDRA-9761)
 + * Cancel transaction for sstables we wont redistribute index summary
 +   for (CASSANDRA-10270)
 + * Handle missing RoleManager in config after upgrade to 2.2 (CASSANDRA-10209) 
 + * Retry snapshot deletion after compaction and gc on Windows (CASSANDRA-10222)
 + * Fix failure to start with space in directory path on Windows (CASSANDRA-10239)
 + * Fix repair hang when snapshot failed (CASSANDRA-10057)
 + * Fall back to 1/4 commitlog volume for commitlog_total_space on small disks
 +   (CASSANDRA-10199)
 +Merged from 2.1:
+  * (Pig) support BulkOutputFormat as a URL parameter (CASSANDRA-7410)
   * BATCH statement is broken in cqlsh (CASSANDRA-10272)
   * Added configurable warning threshold for GC duration (CASSANDRA-8907)
 - * (cqlsh) Make cqlsh PEP8 compliant (CASSANDRA-10066)
 + * (cqlsh) Make cqlsh PEP8 Compliant (CASSANDRA-10066)
   * (cqlsh) Fix error when starting cqlsh with --debug (CASSANDRA-10282)
   * Scrub, Cleanup and Upgrade do not unmark compacting until all operations
     have completed, regardless of the occurence of exceptions (CASSANDRA-10274)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
index 4dd53ff,e8de0f2..9c45bfe
--- a/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
+++ b/src/java/org/apache/cassandra/hadoop/AbstractColumnFamilyInputFormat.java
@@@ -24,23 -32,31 +24,24 @@@ import java.util.concurrent.*
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
 -import org.apache.cassandra.auth.IAuthenticator;
 +import com.datastax.driver.core.Host;
 +import com.datastax.driver.core.Metadata;
 +import com.datastax.driver.core.ResultSet;
 +import com.datastax.driver.core.Row;
 +import com.datastax.driver.core.Session;
 +import com.datastax.driver.core.TokenRange;
++
 +import org.apache.cassandra.db.SystemKeyspace;
 +import org.apache.cassandra.dht.ByteOrderedPartitioner;
  import org.apache.cassandra.dht.IPartitioner;
 +import org.apache.cassandra.dht.OrderPreservingPartitioner;
  import org.apache.cassandra.dht.Range;
  import org.apache.cassandra.dht.Token;
 -import org.apache.cassandra.thrift.AuthenticationRequest;
 -import org.apache.cassandra.thrift.Cassandra;
 -import org.apache.cassandra.thrift.CfSplit;
 -import org.apache.cassandra.thrift.InvalidRequestException;
 +import org.apache.cassandra.hadoop.cql3.*;
  import org.apache.cassandra.thrift.KeyRange;
 -import org.apache.cassandra.thrift.TokenRange;
 -import org.apache.commons.lang3.StringUtils;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.mapred.JobConf;
 -import org.apache.hadoop.mapreduce.InputFormat;
 -import org.apache.hadoop.mapreduce.InputSplit;
 -import org.apache.hadoop.mapreduce.JobContext;
 -import org.apache.hadoop.mapreduce.TaskAttemptContext;
 -import org.apache.hadoop.mapreduce.TaskAttemptID;
 -import org.apache.thrift.TApplicationException;
 -import org.apache.thrift.TException;
 -import org.apache.thrift.protocol.TBinaryProtocol;
 -import org.apache.thrift.protocol.TProtocol;
 -import org.apache.thrift.transport.TTransport;
 -import org.apache.thrift.transport.TTransportException;
 -
 +import org.apache.hadoop.mapreduce.*;
  
  public abstract class AbstractColumnFamilyInputFormat<K, Y> extends InputFormat<K,
Y> implements org.apache.hadoop.mapred.InputFormat<K, Y>
  {
@@@ -226,64 -271,83 +227,65 @@@
          }
      }
  
 -    private List<CfSplit> getSubSplits(String keyspace, String cfName, TokenRange
range, Configuration conf) throws IOException
 +    private Map<TokenRange, Long> getSubSplits(String keyspace, String cfName, TokenRange
range, Configuration conf) throws IOException
      {
 -        int splitsize = ConfigHelper.getInputSplitSize(conf);
 -        for (int i = 0; i < range.rpc_endpoints.size(); i++)
 +        int splitSize = ConfigHelper.getInputSplitSize(conf);
 +        try
          {
 -            String host = range.rpc_endpoints.get(i);
 -
 -            if (host == null || host.equals("0.0.0.0"))
 -                host = range.endpoints.get(i);
 -
 -            try
 -            {
 -                Cassandra.Client client = ConfigHelper.createConnection(conf, host, ConfigHelper.getInputRpcPort(conf));
 -                client.set_keyspace(keyspace);
 -
 -                try
 -                {
 -                    return client.describe_splits_ex(cfName, range.start_token, range.end_token,
splitsize);
 -                }
 -                catch (TApplicationException e)
 -                {
 -                    // fallback to guessing split size if talking to a server without describe_splits_ex
method
 -                    if (e.getType() == TApplicationException.UNKNOWN_METHOD)
 -                    {
 -                        List<String> splitPoints = client.describe_splits(cfName,
range.start_token, range.end_token, splitsize);
 -                        return tokenListToSplits(splitPoints, splitsize);
 -                    }
 -                    throw e;
 -                }
 -            }
 -            catch (IOException e)
 -            {
 -                logger.debug("failed connect to endpoint {}", host, e);
 -            }
 -            catch (InvalidRequestException e)
 -            {
 -                throw new RuntimeException(e);
 -            }
 -            catch (TException e)
 -            {
 -                throw new RuntimeException(e);
 -            }
 +            return describeSplits(keyspace, cfName, range, splitSize);
 +        }
 +        catch (Exception e)
 +        {
 +            throw new RuntimeException(e);
          }
 -        throw new IOException("failed connecting to all endpoints " + StringUtils.join(range.endpoints,
","));
      }
  
 -    private List<CfSplit> tokenListToSplits(List<String> splitTokens, int splitsize)
 +    private Map<TokenRange, Set<Host>> getRangeMap(Configuration conf, String
keyspace)
      {
 -        List<CfSplit> splits = Lists.newArrayListWithExpectedSize(splitTokens.size()
- 1);
 -        for (int j = 0; j < splitTokens.size() - 1; j++)
 -            splits.add(new CfSplit(splitTokens.get(j), splitTokens.get(j + 1), splitsize));
 -        return splits;
 +        try (Session session = CqlConfigHelper.getInputCluster(ConfigHelper.getInputInitialAddress(conf).split(","),
conf).connect())
 +        {
 +            Map<TokenRange, Set<Host>> map = new HashMap<>();
 +            Metadata metadata = session.getCluster().getMetadata();
 +            for (TokenRange tokenRange : metadata.getTokenRanges())
 +                map.put(tokenRange, metadata.getReplicas('"' + keyspace + '"', tokenRange));
 +            return map;
 +        }
      }
  
 -    private List<TokenRange> getRangeMap(Configuration conf) throws IOException
 +    private Map<TokenRange, Long> describeSplits(String keyspace, String table, TokenRange
tokenRange, int splitSize)
      {
 -        Cassandra.Client client = ConfigHelper.getClientFromInputAddressList(conf);
 -
 -        List<TokenRange> map;
 -        try
 +        String query = String.format("SELECT mean_partition_size, partitions_count " +
 +                                     "FROM %s.%s " +
 +                                     "WHERE keyspace_name = ? AND table_name = ? AND range_start
= ? AND range_end = ?",
 +                                     SystemKeyspace.NAME,
 +                                     SystemKeyspace.SIZE_ESTIMATES);
 +
 +        ResultSet resultSet = session.execute(query, keyspace, table, tokenRange.getStart().toString(),
tokenRange.getEnd().toString());
 +
 +        Row row = resultSet.one();
 +        // If we have no data on this split, return the full split i.e., do not sub-split
 +        // Assume smallest granularity of partition count available from CASSANDRA-7688
 +        if (row == null)
          {
 -            map = client.describe_local_ring(ConfigHelper.getInputKeyspace(conf));
 +            Map<TokenRange, Long> wrappedTokenRange = new HashMap<>();
 +            wrappedTokenRange.put(tokenRange, (long) 128);
 +            return wrappedTokenRange;
          }
 -        catch (InvalidRequestException e)
 -        {
 -            throw new RuntimeException(e);
 -        }
 -        catch (TException e)
 -        {
 -            throw new RuntimeException(e);
 -        }
 -        return map;
 +
 +        long meanPartitionSize = row.getLong("mean_partition_size");
 +        long partitionCount = row.getLong("partitions_count");
 +
 +        int splitCount = (int)((meanPartitionSize * partitionCount) / splitSize);
++        if (splitCount <= 0) splitCount = 1;
 +        List<TokenRange> splitRanges = tokenRange.splitEvenly(splitCount);
 +        Map<TokenRange, Long> rangesWithLength = new HashMap<>();
 +        for (TokenRange range : splitRanges)
 +            rangesWithLength.put(range, partitionCount/splitCount);
 +
 +        return rangesWithLength;
      }
  
 -    //
      // Old Hadoop API
 -    //
      public org.apache.hadoop.mapred.InputSplit[] getSplits(JobConf jobConf, int numSplits)
throws IOException
      {
          TaskAttemptContext tac = HadoopCompat.newTaskAttemptContext(jobConf, new TaskAttemptID());

http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
index e77c4c8,ced8aa9..9e6e23b
--- a/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
+++ b/src/java/org/apache/cassandra/hadoop/cql3/CqlBulkRecordWriter.java
@@@ -23,16 -22,15 +23,17 @@@ import java.io.IOException
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  import java.nio.ByteBuffer;
 -import java.util.HashSet;
 -import java.util.List;
 -import java.util.Set;
 -import java.util.UUID;
 +import java.util.*;
 +import java.util.concurrent.*;
  
 -import org.apache.cassandra.config.EncryptionOptions;
 -import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
 +import org.slf4j.Logger;
 +import org.slf4j.LoggerFactory;
- 
 +import org.apache.cassandra.config.CFMetaData;
 +import org.apache.cassandra.config.Config;
 +import org.apache.cassandra.config.DatabaseDescriptor;
++import org.apache.cassandra.dht.IPartitioner;
++import org.apache.cassandra.dht.Murmur3Partitioner;
  import org.apache.cassandra.exceptions.InvalidRequestException;
 -import org.apache.cassandra.hadoop.AbstractBulkRecordWriter;
  import org.apache.cassandra.hadoop.BulkRecordWriter;
  import org.apache.cassandra.hadoop.ConfigHelper;
  import org.apache.cassandra.hadoop.HadoopCompat;
@@@ -84,6 -67,6 +85,7 @@@ public class CqlBulkRecordWriter extend
      private String insertStatement;
      private File outputDir;
      private boolean deleteSrc;
++    private IPartitioner partitioner;
  
      CqlBulkRecordWriter(TaskAttemptContext context) throws IOException
      {
@@@ -113,55 -90,45 +115,64 @@@
      {
          // if anything is missing, exceptions will be thrown here, instead of on write()
          keyspace = ConfigHelper.getOutputKeyspace(conf);
 -        columnFamily = ConfigHelper.getOutputColumnFamily(conf);
 -        schema = CqlBulkOutputFormat.getColumnFamilySchema(conf, columnFamily);
 -        insertStatement = CqlBulkOutputFormat.getColumnFamilyInsertStatement(conf, columnFamily);
 -        outputDir = getColumnFamilyDirectory();
 +        table = ConfigHelper.getOutputColumnFamily(conf);
 +        
 +        // check if table is aliased
 +        String aliasedCf = CqlBulkOutputFormat.getTableForAlias(conf, table);
 +        if (aliasedCf != null)
 +            table = aliasedCf;
 +        
 +        schema = CqlBulkOutputFormat.getTableSchema(conf, table);
 +        insertStatement = CqlBulkOutputFormat.getTableInsertStatement(conf, table);
 +        outputDir = getTableDirectory();
          deleteSrc = CqlBulkOutputFormat.getDeleteSourceOnSuccess(conf);
++        try
++        {
++            partitioner = ConfigHelper.getInputPartitioner(conf);
++        }
++        catch (Exception e)
++        {
++            partitioner = Murmur3Partitioner.instance;
++        }
 +    }
 +
 +    protected String getOutputLocation() throws IOException
 +    {
 +        String dir = conf.get(OUTPUT_LOCATION, System.getProperty("java.io.tmpdir"));
 +        if (dir == null)
 +            throw new IOException("Output directory not defined, if hadoop is not setting
java.io.tmpdir then define " + OUTPUT_LOCATION);
 +        return dir;
      }
  
 -    
      private void prepareWriter() throws IOException
      {
 -        try
 +        if (writer == null)
          {
 -            if (writer == null)
 -            {
 -                writer = CQLSSTableWriter.builder()
 -                    .forTable(schema)
 -                    .using(insertStatement)
 -                    .withPartitioner(ConfigHelper.getOutputPartitioner(conf))
 -                    .inDirectory(outputDir)
 -                    .withBufferSizeInMB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")))
 -                    .build();
 -            }
 -            if (loader == null)
 -            {
 -                BulkLoader.ExternalClient externalClient = getExternalClient(conf);
 -                this.loader = new SSTableLoader(outputDir, externalClient, new BulkRecordWriter.NullOutputHandler())
{
 -                    @Override
 -                    public void onSuccess(StreamState finalState)
 -                    {
 -                        if (deleteSrc)
 -                            FileUtils.deleteRecursive(outputDir);
 -                    }
 -                };
 -            }
 +            writer = CQLSSTableWriter.builder()
 +                                     .forTable(schema)
 +                                     .using(insertStatement)
 +                                     .withPartitioner(ConfigHelper.getOutputPartitioner(conf))
 +                                     .inDirectory(outputDir)
 +                                     .withBufferSizeInMB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB,
"64")))
++                                     .withPartitioner(partitioner)
 +                                     .build();
          }
 -        catch (Exception e)
 +
 +        if (loader == null)
          {
 -            throw new IOException(e);
 -        }      
 +            ExternalClient externalClient = new ExternalClient(conf);
 +            externalClient.setTableMetadata(CFMetaData.compile(schema, keyspace));
 +
 +            loader = new SSTableLoader(outputDir, externalClient, new BulkRecordWriter.NullOutputHandler())
 +            {
 +                @Override
 +                public void onSuccess(StreamState finalState)
 +                {
 +                    if (deleteSrc)
 +                        FileUtils.deleteRecursive(outputDir);
 +                }
 +            };
 +        }
      }
      
      /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
index dc3c174,5287bf5..223a848
--- a/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
+++ b/src/java/org/apache/cassandra/hadoop/pig/CqlNativeStorage.java
@@@ -17,71 -17,47 +17,75 @@@
   */
  package org.apache.cassandra.hadoop.pig;
  
 +import java.io.ByteArrayInputStream;
 +import java.io.ByteArrayOutputStream;
  import java.io.IOException;
 +import java.io.ObjectInputStream;
 +import java.io.ObjectOutputStream;
 +import java.io.Serializable;
 +import java.io.UnsupportedEncodingException;
 +import java.net.URLDecoder;
  import java.nio.ByteBuffer;
 -import java.nio.charset.CharacterCodingException;
  import java.util.*;
  
 -import org.apache.cassandra.config.CFMetaData;
 -import org.apache.cassandra.config.ColumnDefinition;
 -import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption;
 -import org.apache.cassandra.db.BufferCell;
 -import org.apache.cassandra.db.Cell;
 -import org.apache.cassandra.db.composites.CellNames;
 +import com.datastax.driver.core.ColumnMetadata;
 +import com.datastax.driver.core.Metadata;
 +import com.datastax.driver.core.Row;
 +import com.datastax.driver.core.Session;
 +import com.datastax.driver.core.TableMetadata;
 +import com.datastax.driver.core.exceptions.NoHostAvailableException;
++
  import org.apache.cassandra.db.marshal.*;
 +import org.apache.cassandra.exceptions.AuthenticationException;
  import org.apache.cassandra.exceptions.ConfigurationException;
 -import org.apache.cassandra.hadoop.AbstractBulkRecordWriter;
 +import org.apache.cassandra.exceptions.SyntaxException;
  import org.apache.cassandra.hadoop.ConfigHelper;
  import org.apache.cassandra.hadoop.HadoopCompat;
+ import org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat;
++import org.apache.cassandra.hadoop.cql3.CqlBulkRecordWriter;
  import org.apache.cassandra.hadoop.cql3.CqlConfigHelper;
  import org.apache.cassandra.hadoop.cql3.CqlRecordReader;
 -import org.apache.cassandra.thrift.*;
 +import org.apache.cassandra.serializers.CollectionSerializer;
  import org.apache.cassandra.utils.*;
 -import org.apache.commons.lang.StringUtils;
 +import org.apache.hadoop.conf.Configuration;
 +import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.mapreduce.*;
 -import org.apache.pig.Expression;
 -import org.apache.pig.ResourceSchema;
 +import org.apache.pig.*;
  import org.apache.pig.Expression.OpType;
  import org.apache.pig.ResourceSchema.ResourceFieldSchema;
  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.impl.util.UDFContext;
 -import org.apache.thrift.TException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
 +import org.yaml.snakeyaml.external.biz.base64Coder.Base64Coder;
  
 -import com.datastax.driver.core.Row;
  
 -public class CqlNativeStorage extends AbstractCassandraStorage
 +public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, LoadMetadata
  {
 +    protected String DEFAULT_INPUT_FORMAT;
 +    protected String DEFAULT_OUTPUT_FORMAT;
 +
 +    protected String username;
 +    protected String password;
 +    protected String keyspace;
 +    protected String column_family;
 +    protected String loadSignature;
 +    protected String storeSignature;
 +
 +    protected Configuration conf;
 +    protected String inputFormatClass;
 +    protected String outputFormatClass;
 +    protected int splitSize = 64 * 1024;
 +    protected String partitionerClass;
 +    protected boolean usePartitionFilter = false;
 +    protected String initHostAddress;
 +    protected String rpcPort;
 +    protected int nativeProtocolVersion = 1;
 +
      private static final Logger logger = LoggerFactory.getLogger(CqlNativeStorage.class);
 -    public static String BULK_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat";
++    private static String BULK_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat";
      private int pageSize = 1000;
      private String columns;
      private String outputQuery;
@@@ -109,6 -87,22 +113,16 @@@
      private String nativeSSLCipherSuites;
      private String inputCql;
  
+     private boolean bulkOutputFormat = false;
+     private String bulkCfSchema;
+     private String bulkInsertStatement;
+     private String bulkOutputLocation;
+     private int bulkBuffSize = -1;
+     private int bulkStreamThrottle = -1;
+     private int bulkMaxFailedHosts = -1;
 -    private int storagePort = CqlBulkOutputFormat.DEFAULT_STORAGE_PORT;
 -    private int sslStoragePort = CqlBulkOutputFormat.DEFAULT_SSL_STORAGE_PORT;
 -    private String serverKeystore;
 -    private String serverKeystorePassword;
 -    private String serverTruststore;
 -    private String serverTruststorePassword;
 -    private String serverCipherSuites;
 -    private String internodeEncrypt;
++    private boolean bulkDeleteSourceOnSuccess = true;
++    private String bulkTableAlias;
+ 
      public CqlNativeStorage()
      {
          this(1000);
@@@ -241,83 -240,188 +255,48 @@@
          return obj;
      }
  
 -    /** include key columns */
 -    protected List<ColumnDef> getColumnMetadata(Cassandra.Client client)
 -            throws InvalidRequestException,
 -            UnavailableException,
 -            TimedOutException,
 -            SchemaDisagreementException,
 -            TException,
 -            CharacterCodingException,
 -            org.apache.cassandra.exceptions.InvalidRequestException,
 -            ConfigurationException,
 -            NotFoundException
 -    {
 -        List<ColumnDef> keyColumns = null;
 -        // get key columns
 +    /** get the columnfamily definition for the signature */
 +    protected TableInfo getCfInfo(String signature) throws IOException
 +    {
 +        UDFContext context = UDFContext.getUDFContext();
 +        Properties property = context.getUDFProperties(CqlNativeStorage.class);
 +        TableInfo cfInfo;
          try
          {
 -            keyColumns = getKeysMeta(client);
 +            cfInfo = cfdefFromString(property.getProperty(signature));
          }
 -        catch(Exception e)
 +        catch (ClassNotFoundException e)
          {
 -            logger.error("Error in retrieving key columns" , e);
 +            throw new IOException(e);
          }
 -
 -        // get other columns
 -        List<ColumnDef> columns = getColumnMeta(client, false, !hasCompactValueAlias);
 -
 -        // combine all columns in a list
 -        if (keyColumns != null && columns != null)
 -            keyColumns.addAll(columns);
 -
 -        return keyColumns;
 +        return cfInfo;
      }
  
 -    /** get keys meta data */
 -    private List<ColumnDef> getKeysMeta(Cassandra.Client client)
 -            throws Exception
 +    /** return the CfInfo for the column family */
 +    protected TableMetadata getCfInfo(Session client)
 +            throws NoHostAvailableException,
 +            AuthenticationException,
 +            IllegalStateException
      {
 -        String query = "SELECT key_aliases, " +
 -                "       column_aliases, " +
 -                "       key_validator, " +
 -                "       comparator, " +
 -                "       keyspace_name, " +
 -                "       value_alias, " +
 -                "       default_validator " +
 -                "FROM system.schema_columnfamilies " +
 -                "WHERE keyspace_name = '%s'" +
 -                "  AND columnfamily_name = '%s' ";
 -
 -        CqlResult result = client.execute_cql3_query(
 -                ByteBufferUtil.bytes(String.format(query, keyspace, column_family)),
 -                Compression.NONE,
 -                ConsistencyLevel.ONE);
 -
 -        if (result == null || result.rows == null || result.rows.isEmpty())
 -            return null;
 -
 -        Iterator<CqlRow> iteraRow = result.rows.iterator();
 -        List<ColumnDef> keys = new ArrayList<ColumnDef>();
 -        if (iteraRow.hasNext())
 -        {
 -            CqlRow cqlRow = iteraRow.next();
 -            String name = ByteBufferUtil.string(cqlRow.columns.get(4).value);
 -            logger.debug("Found ksDef name: {}", name);
 -            String keyString = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(0).getValue()));
 -
 -            logger.debug("partition keys: {}", keyString);
 -            List<String> keyNames = FBUtilities.fromJsonList(keyString);
 -
 -            Iterator<String> iterator = keyNames.iterator();
 -            while (iterator.hasNext())
 -            {
 -                ColumnDef cDef = new ColumnDef();
 -                cDef.name = ByteBufferUtil.bytes(iterator.next());
 -                keys.add(cDef);
 -            }
 -            // classic thrift tables
 -            if (keys.size() == 0)
 -            {
 -                CFMetaData cfm = getCFMetaData(keyspace, column_family, client);
 -                for (ColumnDefinition def : cfm.partitionKeyColumns())
 -                {
 -                    String key = def.name.toString();
 -                    logger.debug("name: {} ", key);
 -                    ColumnDef cDef = new ColumnDef();
 -                    cDef.name = ByteBufferUtil.bytes(key);
 -                    keys.add(cDef);
 -                }
 -                for (ColumnDefinition def : cfm.clusteringColumns())
 -                {
 -                    String key = def.name.toString();
 -                    logger.debug("name: {} ", key);
 -                    ColumnDef cDef = new ColumnDef();
 -                    cDef.name = ByteBufferUtil.bytes(key);
 -                    keys.add(cDef);
 -                }
 -            }
 -
 -            keyString = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(1).getValue()));
 -
 -            logger.debug("cluster keys: {}", keyString);
 -            keyNames = FBUtilities.fromJsonList(keyString);
 -
 -            iterator = keyNames.iterator();
 -            while (iterator.hasNext())
 -            {
 -                ColumnDef cDef = new ColumnDef();
 -                cDef.name = ByteBufferUtil.bytes(iterator.next());
 -                keys.add(cDef);
 -            }
 -
 -            String validator = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(2).getValue()));
 -            logger.debug("row key validator: {}", validator);
 -            AbstractType<?> keyValidator = parseType(validator);
 -
 -            Iterator<ColumnDef> keyItera = keys.iterator();
 -            if (keyValidator instanceof CompositeType)
 -            {
 -                Iterator<AbstractType<?>> typeItera = ((CompositeType) keyValidator).types.iterator();
 -                while (typeItera.hasNext())
 -                    keyItera.next().validation_class = typeItera.next().toString();
 -            }
 -            else
 -                keyItera.next().validation_class = keyValidator.toString();
 -
 -            validator = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(3).getValue()));
 -            logger.debug("cluster key validator: {}", validator);
 -
 -            if (keyItera.hasNext() && validator != null && !validator.isEmpty())
 -            {
 -                AbstractType<?> clusterKeyValidator = parseType(validator);
 -
 -                if (clusterKeyValidator instanceof CompositeType)
 -                {
 -                    Iterator<AbstractType<?>> typeItera = ((CompositeType) clusterKeyValidator).types.iterator();
 -                    while (keyItera.hasNext())
 -                        keyItera.next().validation_class = typeItera.next().toString();
 -                }
 -                else
 -                    keyItera.next().validation_class = clusterKeyValidator.toString();
 -            }
 -
 -            // compact value_alias column
 -            if (cqlRow.columns.get(5).value != null)
 -            {
 -                try
 -                {
 -                    String compactValidator = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(6).getValue()));
 -                    logger.debug("default validator: {}", compactValidator);
 -                    AbstractType<?> defaultValidator = parseType(compactValidator);
 -
 -                    ColumnDef cDef = new ColumnDef();
 -                    cDef.name = cqlRow.columns.get(5).value;
 -                    cDef.validation_class = defaultValidator.toString();
 -                    keys.add(cDef);
 -                    hasCompactValueAlias = true;
 -                }
 -                catch (Exception e)
 -                {
 -                    // no compact column at value_alias
 -                }
 -            }
 -
 -        }
 -        return keys;
 +        // get CF meta data
 +        return client.getCluster().getMetadata().getKeyspace(Metadata.quote(keyspace)).getTable(Metadata.quote(column_family));
      }
  
-     /** output: (((name, value), (name, value)), (value ... value), (value...value)) */
-     public void putNext(Tuple t) throws IOException
-     {
-         if (t.size() < 1)
-         {
-             // simply nothing here, we can't even delete without a key
-             logger.warn("Empty output skipped, filter empty tuples to suppress this warning");
-             return;
-         }
- 
-         if (t.getType(0) == DataType.TUPLE)
-         {
-             if (t.getType(1) == DataType.TUPLE)
-             {
-                 Map<String, ByteBuffer> key = tupleToKeyMap((Tuple)t.get(0));
-                 cqlQueryFromTuple(key, t, 1);
-             }
-             else
-                 throw new IOException("Second argument in output must be a tuple");
-         }
-         else
-             throw new IOException("First argument in output must be a tuple");
-     }
- 
      /** convert key tuple to key map */
      private Map<String, ByteBuffer> tupleToKeyMap(Tuple t) throws IOException
      {
          Map<String, ByteBuffer> keys = new HashMap<String, ByteBuffer>();
          for (int i = 0; i < t.size(); i++)
          {
-             if (t.getType(i) == DataType.TUPLE)
-             {
-                 Tuple inner = (Tuple) t.get(i);
-                 if (inner.size() == 2)
-                 {
-                     Object name = inner.get(0);
-                     if (name != null)
-                     {
-                         keys.put(name.toString(), objToBB(inner.get(1)));
-                     }
-                     else
-                         throw new IOException("Key name was empty");
-                 }
-                 else
-                     throw new IOException("Keys were not in name and value pairs");
-             }
-             else
-             {
+             if (t.getType(i) != DataType.TUPLE)
                  throw new IOException("keys was not a tuple");
-             }
 -
+             Tuple inner = (Tuple) t.get(i);
+             if (inner.size() != 2)
+                 throw new IOException("Keys were not in name and value pairs");
+             Object name = inner.get(0);
+             if (name == null)
+                 throw new IOException("Key name was empty");
+             keys.put(name.toString(), objToBB(inner.get(1)));
          }
          return keys;
      }
@@@ -543,10 -537,41 +517,41 @@@
      private String getWhereClauseForPartitionFilter()
      {
          UDFContext context = UDFContext.getUDFContext();
 -        Properties property = context.getUDFProperties(AbstractCassandraStorage.class);
 -        return property.getProperty(PARTITION_FILTER_SIGNATURE);
 +        Properties property = context.getUDFProperties(CqlNativeStorage.class);
 +        return property.getProperty(StorageHelper.PARTITION_FILTER_SIGNATURE);
      }
  
+     /**
+      *  output: (((name, value), (name, value)), (value ... value), (value...value))
+      *  bulk output: ((value ... value), (value...value))
+      *
+      * */
+     public void putNext(Tuple t) throws IOException
+     {
+         if (t.size() < 1)
+         {
+             // simply nothing here, we can't even delete without a key
+             logger.warn("Empty output skipped, filter empty tuples to suppress this warning");
+             return;
+         }
+ 
+         if (t.getType(0) != DataType.TUPLE)
+             throw new IOException("First argument in output must be a tuple");
+ 
+         if (!bulkOutputFormat && t.getType(1) != DataType.TUPLE)
+             throw new IOException("Second argument in output must be a tuple");
+ 
+         if (bulkOutputFormat)
+         {
+             cqlQueryFromTuple(null, t, 0);
+         }
+         else
+         {
+             Map<String, ByteBuffer> key = tupleToKeyMap((Tuple)t.get(0));
+             cqlQueryFromTuple(key, t, 1);
+         }
+     }
+ 
      /** set read configuration settings */
      public void setLocation(String location, Job job) throws IOException
      {
@@@ -672,6 -699,42 +677,32 @@@
          ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
          CqlConfigHelper.setOutputCql(conf, outputQuery);
  
+         if (bulkOutputFormat)
+         {
+             DEFAULT_OUTPUT_FORMAT = BULK_OUTPUT_FORMAT;
+             if (bulkCfSchema != null)
 -                CqlBulkOutputFormat.setColumnFamilySchema(conf, column_family, bulkCfSchema);
++                CqlBulkOutputFormat.setTableSchema(conf, column_family, bulkCfSchema);
+             else
+                 throw new IOException("bulk_cf_schema is missing in input url parameter");
+             if (bulkInsertStatement != null)
 -                CqlBulkOutputFormat.setColumnFamilyInsertStatement(conf, column_family,
bulkInsertStatement);
++                CqlBulkOutputFormat.setTableInsertStatement(conf, column_family, bulkInsertStatement);
+             else
+                 throw new IOException("bulk_insert_statement is missing in input url parameter");
++            if (bulkTableAlias != null)
++                CqlBulkOutputFormat.setTableAlias(conf, bulkTableAlias, column_family);

++            CqlBulkOutputFormat.setDeleteSourceOnSuccess(conf, bulkDeleteSourceOnSuccess);
+             if (bulkOutputLocation != null)
 -                conf.set(AbstractBulkRecordWriter.OUTPUT_LOCATION, bulkOutputLocation);
++                conf.set(CqlBulkRecordWriter.OUTPUT_LOCATION, bulkOutputLocation);
+             if (bulkBuffSize > 0)
 -                conf.set(AbstractBulkRecordWriter.BUFFER_SIZE_IN_MB, String.valueOf(bulkBuffSize));
++                conf.set(CqlBulkRecordWriter.BUFFER_SIZE_IN_MB, String.valueOf(bulkBuffSize));
+             if (bulkStreamThrottle > 0)
 -                conf.set(AbstractBulkRecordWriter.STREAM_THROTTLE_MBITS, String.valueOf(bulkStreamThrottle));
++                conf.set(CqlBulkRecordWriter.STREAM_THROTTLE_MBITS, String.valueOf(bulkStreamThrottle));
+             if (bulkMaxFailedHosts > 0)
 -                conf.set(AbstractBulkRecordWriter.MAX_FAILED_HOSTS, String.valueOf(bulkMaxFailedHosts));
 -            CqlBulkOutputFormat.setSSLStoragePort(conf, sslStoragePort);
 -            CqlBulkOutputFormat.setStoragePort(conf, storagePort);
 -            if (serverEncrypted())
 -            {
 -                if (!StringUtils.isEmpty(serverKeystore))
 -                    CqlBulkOutputFormat.setServerKeystore(conf, serverKeystore);
 -                if (!StringUtils.isEmpty(serverTruststore))
 -                    CqlBulkOutputFormat.setServerTruststore(conf, serverTruststore);
 -                if (!StringUtils.isEmpty(serverKeystorePassword))
 -                    CqlBulkOutputFormat.setServerKeystorePassword(conf, serverKeystorePassword);
 -                if (!StringUtils.isEmpty(serverTruststorePassword))
 -                    CqlBulkOutputFormat.setServerTruststorePassword(conf, serverTruststorePassword);
 -                if (!StringUtils.isEmpty(serverCipherSuites))
 -                    CqlBulkOutputFormat.setServerCipherSuites(conf, serverCipherSuites);
 -            }
++                conf.set(CqlBulkRecordWriter.MAX_FAILED_HOSTS, String.valueOf(bulkMaxFailedHosts));
++            if (partitionerClass!= null)
++                ConfigHelper.setInputPartitioner(conf, partitionerClass);
+         }
+ 
          setConnectionInformation();
  
          if (ConfigHelper.getOutputRpcPort(conf) == 0)
@@@ -773,6 -773,37 +804,25 @@@
                  if (urlQuery.containsKey("output_query"))
                      outputQuery = urlQuery.get("output_query");
  
+                 if (urlQuery.containsKey("bulk_output_format"))
+                     bulkOutputFormat = Boolean.valueOf(urlQuery.get("bulk_output_format"));
+                 if (urlQuery.containsKey("bulk_cf_schema"))
+                     bulkCfSchema = urlQuery.get("bulk_cf_schema");
+                 if (urlQuery.containsKey("bulk_insert_statement"))
+                     bulkInsertStatement = urlQuery.get("bulk_insert_statement");
+                 if (urlQuery.containsKey("bulk_output_location"))
+                     bulkOutputLocation = urlQuery.get("bulk_output_location");
+                 if (urlQuery.containsKey("bulk_buff_size"))
+                     bulkBuffSize = Integer.valueOf(urlQuery.get("bulk_buff_size"));
+                 if (urlQuery.containsKey("bulk_stream_throttle"))
+                     bulkStreamThrottle = Integer.valueOf(urlQuery.get("bulk_stream_throttle"));
+                 if (urlQuery.containsKey("bulk_max_failed_hosts"))
+                     bulkMaxFailedHosts = Integer.valueOf(urlQuery.get("bulk_max_failed_hosts"));
 -                if (urlQuery.containsKey("storage_port"))
 -                    storagePort = Integer.valueOf(urlQuery.get("storage_port"));
 -                if (urlQuery.containsKey("ssl_storage_port"))
 -                    sslStoragePort = Integer.valueOf(urlQuery.get("ssl_storage_port"));
 -                if (urlQuery.containsKey("internode_encrypt"))
 -                    internodeEncrypt = urlQuery.get("internode_encrypt");
 -                if (urlQuery.containsKey("server_keystore"))
 -                    serverKeystore = urlQuery.get("server_keystore");
 -                if (urlQuery.containsKey("server_truststore"))
 -                    serverTruststore = urlQuery.get("server_truststore");
 -                if (urlQuery.containsKey("server_keystore_pass"))
 -                    serverKeystorePassword = urlQuery.get("server_keystore_pass");
 -                if (urlQuery.containsKey("server_truststore_pass"))
 -                    serverTruststorePassword = urlQuery.get("server_truststore_pass");
 -                if (urlQuery.containsKey("server_cipher_suites"))
 -                    serverCipherSuites = urlQuery.get("server_cipher_suites");
++                if (urlQuery.containsKey("bulk_delete_source"))
++                    bulkDeleteSourceOnSuccess = Boolean.parseBoolean(urlQuery.get("bulk_delete_source"));
++                if (urlQuery.containsKey("bulk_table_alias"))
++                    bulkTableAlias = urlQuery.get("bulk_table_alias");
+ 
                  //split size
                  if (urlQuery.containsKey("split_size"))
                      splitSize = Integer.parseInt(urlQuery.get("split_size"));
@@@ -855,10 -888,20 +905,13 @@@
                      "[&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>]"
+
-                     "[columns=<columns>][where_clause=<where_clause>]]': " +
e.getMessage());
-         }
+                     "[columns=<columns>][where_clause=<where_clause>]" +
 -                    "[&bulk_cf_schema=bulk_cf_schema][&bulk_insert_statement=bulk_insert_statement]"
+
 -                    "[&bulk_output_location=<bulk_output_location>][&bulk_buff_size=<bulk_buff_size>]"
+
 -                    "[&storage_port=<storage_port>][&ssl_storage_port=<ssl_storage_port>]"
+
 -                    "[&server_keystore=<server_keystore>][&server_keystore_pass=<server_keystore_pass>]"
+
 -                    "[&server_truststore=<server_truststore>][&server_truststore_pass=<server_truststore_pass>]"
+
 -                    "[&server_cipher_suites=<server_cipher_suites>][&internode_encrypt=<internode_encrypt>]"
+
++                    "[&bulk_cf_schema=bulk_cf_schema][&bulk_insert_statement=bulk_insert_statement][&bulk_table_alias=<bulk_table_alias>]"
+
++                    "[&bulk_output_location=<bulk_output_location>][&bulk_buff_size=<bulk_buff_size>][&bulk_delete_source=<bulk_delete_source>]"
+
+                     "[&bulk_stream_throttle=<bulk_stream_throttle>][&bulk_max_failed_hosts=<bulk_max_failed_hosts>]]':
" +  e.getMessage());
+          }
      }
  
 -    /**
 -     * Thrift API can't handle null, so use empty byte array
 -     */
      public ByteBuffer nullToBB()
      {
          return ByteBuffer.wrap(new byte[0]);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/test/conf/cassandra_pig.yaml
----------------------------------------------------------------------
diff --cc test/conf/cassandra_pig.yaml
index 0000000,0000000..68615cf
new file mode 100644
--- /dev/null
+++ b/test/conf/cassandra_pig.yaml
@@@ -1,0 -1,0 +1,41 @@@
++#
++# Warning!
++# Consider the effects on 'o.a.c.i.s.LegacySSTableTest' before changing schemas in this
file.
++#
++cluster_name: Test Cluster
++memtable_allocation_type: offheap_objects
++commitlog_sync: batch
++commitlog_sync_batch_window_in_ms: 1.0
++commitlog_segment_size_in_mb: 5
++commitlog_directory: build/test/cassandra/commitlog
++partitioner: org.apache.cassandra.dht.Murmur3Partitioner
++listen_address: 127.0.0.1
++storage_port: 7010
++rpc_port: 9170
++start_native_transport: true
++native_transport_port: 9042
++column_index_size_in_kb: 4
++saved_caches_directory: build/test/cassandra/saved_caches
++data_file_directories:
++    - build/test/cassandra/data
++disk_access_mode: mmap
++seed_provider:
++    - class_name: org.apache.cassandra.locator.SimpleSeedProvider
++      parameters:
++          - seeds: "127.0.0.1"
++endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch
++dynamic_snitch: true
++request_scheduler: org.apache.cassandra.scheduler.RoundRobinScheduler
++request_scheduler_id: keyspace
++server_encryption_options:
++    internode_encryption: none
++    keystore: conf/.keystore
++    keystore_password: cassandra
++    truststore: conf/.truststore
++    truststore_password: cassandra
++incremental_backups: true
++concurrent_compactors: 4
++compaction_throughput_mb_per_sec: 0
++row_cache_class_name: org.apache.cassandra.cache.OHCProvider
++row_cache_size_in_mb: 16
++enable_user_defined_functions: true

http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/test/pig/org/apache/cassandra/pig/CqlTableTest.java
----------------------------------------------------------------------
diff --cc test/pig/org/apache/cassandra/pig/CqlTableTest.java
index 72fdd5a,2e1758e..3902fce
--- a/test/pig/org/apache/cassandra/pig/CqlTableTest.java
+++ b/test/pig/org/apache/cassandra/pig/CqlTableTest.java
@@@ -265,4 -317,33 +272,32 @@@ public class CqlTableTest extends PigTe
              Assert.fail("Can't fetch any data");
          }
      }
+ 
+     @Test
 -    public void testCqlStorageSingleKeyTableBulkLoad()
 -    throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException,
TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException
++    public void testCqlStorageSingleKeyTableBulkLoad() throws TException, IOException
+     {
+         pig.setBatchOn();
+         //input_cql=select * from moredata where token(x) > ? and token(x) <= ?
+         pig.registerQuery("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();");
+         pig.registerQuery("insertformat= FOREACH moretestvalues GENERATE TOTUPLE(x, y);");
+         pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/test_bulk?" + defaultParameters
+ nativeParameters +  "&bulk_output_format=true&bulk_cf_schema=CREATE%20TABLE%20cql3ks.test_bulk%20(a%20int%20PRIMARY%20KEY%2C%20b%20int)&bulk_insert_statement=Insert%20into%20cql3ks.test_bulk(a%2C%20b)%20values(%3F%2C%3F)'
USING CqlNativeStorage();");
+         pig.executeBatch();
+ 
+         //(5,5)
+         //(6,6)
+         //(4,4)
+         //(2,2)
+         //(3,3)
+         //(1,1)
+         //input_cql=select * from test_bulk1 where token(a) > ? and token(a) <= ?
+         pig.registerQuery("result= LOAD 'cql://cql3ks/test_bulk?" + defaultParameters +
nativeParameters + "&input_cql=select%20*%20from%20test_bulk%20where%20token(a)%20%3E%20%3F%20and%20token(a)%20%3C%3D%20%3F'
USING CqlNativeStorage();");
+         Iterator<Tuple> it = pig.openIterator("result");
+         int count = 0;
+         while (it.hasNext()) {
+             Tuple t = it.next();
+             Assert.assertEquals(t.get(0), t.get(1));
+             count ++;
+         }
+         Assert.assertEquals(6, count);
+      }
  }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/cd4a1e6a/test/pig/org/apache/cassandra/pig/PigTestBase.java
----------------------------------------------------------------------
diff --cc test/pig/org/apache/cassandra/pig/PigTestBase.java
index 8c27f6c,e6964f8..a8a9de5
--- a/test/pig/org/apache/cassandra/pig/PigTestBase.java
+++ b/test/pig/org/apache/cassandra/pig/PigTestBase.java
@@@ -54,7 -65,8 +54,7 @@@ public class PigTestBase extends Schema
      protected static Configuration conf;
      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 defaultParameters= "init_address=localhost&rpc_port=9170&partitioner=org.apache.cassandra.dht.ByteOrderedPartitioner"
+
 -                                               "&storage_port=7010&ssl_storage_port=7011&internode_encrypt=NONE";
++    protected static String defaultParameters= "init_address=localhost&rpc_port=9170&partitioner=org.apache.cassandra.dht.Murmur3Partitioner";
      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=9042";
@@@ -62,6 -74,6 +62,7 @@@
      static
      {
          System.setProperty("logback.configurationFile", "logback-test.xml");
++        System.setProperty("cassandra.config", "cassandra_pig.yaml");
      }
  
      @AfterClass


Mime
View raw message