Merge branch 'cassandra-2.2' into cassandra-3.0

CHANGES.txt

@@ -10,6 +10,7 @@ Merged from 2.2:
  * Defer default role manager setup until all nodes are on 2.2+ (CASSANDRA-9761)
  * Handle missing RoleManager in config after upgrade to 2.2 (CASSANDRA-10209)
 Merged from 2.1:
+ * (Pig) support BulkOutputFormat as a URL parameter (CASSANDRA-7410)
  * BATCH statement is broken in cqlsh (CASSANDRA-10272)
  * (cqlsh) Make cqlsh PEP8 Compliant (CASSANDRA-10066)
  * (cqlsh) Fix error when starting cqlsh with --debug (CASSANDRA-10282)

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

@@ -28,10 +28,11 @@ import java.util.concurrent.*;
 
 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.ConfigHelper;
 import org.apache.cassandra.hadoop.HadoopCompat;
@@ -84,6 +85,7 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
     private String insertStatement;
     private File outputDir;
     private boolean deleteSrc;
+    private IPartitioner partitioner;
 
     CqlBulkRecordWriter(TaskAttemptContext context) throws IOException
     {
@@ -124,6 +126,14 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
         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
@@ -144,6 +154,7 @@ public class CqlBulkRecordWriter extends RecordWriter<Object, List<ByteBuffer>>
                                      .withPartitioner(ConfigHelper.getOutputPartitioner(conf))
                                      .inDirectory(outputDir)
                                      .withBufferSizeInMB(Integer.parseInt(conf.get(BUFFER_SIZE_IN_MB, "64")))
+                                     .withPartitioner(partitioner)
                                      .build();
         }
 

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

@@ -34,12 +34,15 @@ 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.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.serializers.CollectionSerializer;
@@ -82,6 +85,7 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
     protected int nativeProtocolVersion = 1;
 
     private static final Logger logger = LoggerFactory.getLogger(CqlNativeStorage.class);
+    private static String BULK_OUTPUT_FORMAT = "org.apache.cassandra.hadoop.cql3.CqlBulkOutputFormat";
     private int pageSize = 1000;
     private String columns;
     private String outputQuery;
@@ -109,6 +113,16 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
     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 boolean bulkDeleteSourceOnSuccess = true;
+    private String bulkTableAlias;
+
     public CqlNativeStorage()
     {
         this(1000);
@@ -268,56 +282,21 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
         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)
+            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
-            {
                 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;
     }
@@ -412,21 +391,16 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
     {
         for (int i = offset; i < t.size(); i++)
         {
-            if (t.getType(i) == DataType.TUPLE)
+            if (t.getType(i) != DataType.TUPLE)
-            {
-                Tuple inner = (Tuple) t.get(i);
-                if (inner.size() > 0)
-                {
-                    List<ByteBuffer> bindedVariables = bindedVariablesFromTuple(inner);
-                    if (bindedVariables.size() > 0)
-                        sendCqlQuery(key, bindedVariables);
-                    else
-                        throw new IOException("Missing binded variables");
-                }
-            }
-            else
-            {
                 throw new IOException("Output type was not a tuple");
+
+            Tuple inner = (Tuple) t.get(i);
+            if (inner.size() > 0)
+            {
+                List<ByteBuffer> bindedVariables = bindedVariablesFromTuple(inner);
+                if (bindedVariables.size() <= 0)
+                    throw new IOException("Missing binded variables");
+                sendCqlQuery(key, bindedVariables);
             }
         }
     }
@@ -547,6 +521,37 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
         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 +677,32 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
         ConfigHelper.setOutputColumnFamily(conf, keyspace, column_family);
         CqlConfigHelper.setOutputCql(conf, outputQuery);
 
+        if (bulkOutputFormat)
+        {
+            DEFAULT_OUTPUT_FORMAT = BULK_OUTPUT_FORMAT;
+            if (bulkCfSchema != null)
+                CqlBulkOutputFormat.setTableSchema(conf, column_family, bulkCfSchema);
+            else
+                throw new IOException("bulk_cf_schema is missing in input url parameter");
+            if (bulkInsertStatement != null)
+                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(CqlBulkRecordWriter.OUTPUT_LOCATION, bulkOutputLocation);
+            if (bulkBuffSize > 0)
+                conf.set(CqlBulkRecordWriter.BUFFER_SIZE_IN_MB, String.valueOf(bulkBuffSize));
+            if (bulkStreamThrottle > 0)
+                conf.set(CqlBulkRecordWriter.STREAM_THROTTLE_MBITS, String.valueOf(bulkStreamThrottle));
+            if (bulkMaxFailedHosts > 0)
+                conf.set(CqlBulkRecordWriter.MAX_FAILED_HOSTS, String.valueOf(bulkMaxFailedHosts));
+            if (partitionerClass!= null)
+                ConfigHelper.setInputPartitioner(conf, partitionerClass);
+        }
+
         setConnectionInformation();
 
         if (ConfigHelper.getOutputRpcPort(conf) == 0)
@@ -773,6 +804,25 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
                 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("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,8 +905,11 @@ public class CqlNativeStorage extends LoadFunc implements StoreFuncInterface, Lo
                     "[&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_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());
+         }
     }
 
     public ByteBuffer nullToBB()

test/conf/cassandra_pig.yaml

@@ -0,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

test/pig/org/apache/cassandra/pig/CqlTableTest.java

@@ -44,6 +44,10 @@ public class CqlTableTest extends PigTestBase
             "CREATE INDEX test_b on test (b);",
 
             "CREATE TABLE moredata (x int PRIMARY KEY, y int);",
+            "CREATE TABLE test_bulk (a int PRIMARY KEY, b int);",
+            "INSERT INTO test_bulk (a,b) VALUES (1,1);",
+            "INSERT INTO test_bulk (a,b) VALUES (2,2);",
+            "INSERT INTO test_bulk (a,b) VALUES (3,3);",
             "INSERT INTO test (a,b) VALUES (1,1);",
             "INSERT INTO test (a,b) VALUES (2,2);",
             "INSERT INTO test (a,b) VALUES (3,3);",
@@ -152,10 +156,13 @@ public class CqlTableTest extends PigTestBase
         //input_cql=select * from test where token(a) > ? and token(a) <= ?
         pig.registerQuery("result= LOAD 'cql://cql3ks/test?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20test%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);
     }
 
     @Test
@@ -265,4 +272,32 @@ public class CqlTableTest extends PigTestBase
             Assert.fail("Can't fetch any data");
         }
     }
+
+    @Test
+    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);
+     }
 }

test/pig/org/apache/cassandra/pig/PigTestBase.java

@@ -46,7 +46,7 @@ public class PigTestBase extends SchemaLoader
     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.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";
@@ -54,6 +54,7 @@ public class PigTestBase extends SchemaLoader
     static
     {
         System.setProperty("logback.configurationFile", "logback-test.xml");
+        System.setProperty("cassandra.config", "cassandra_pig.yaml");
     }
 
     @AfterClass