mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-2.1' into trunk
Conflicts: CHANGES.txt
This commit is contained in:
Brandon Williams
commit
aeea401eef
|
|
@ -15,6 +15,7 @@
|
|||
|
||||
|
||||
2.1.1
|
||||
* Pig support for hadoop CqlInputFormat (CASSANDRA-6454)
|
||||
* Add duration mode to cassandra-stress (CASSANDRA-7468)
|
||||
* Add listen_interface and rpc_interface options (CASSANDRA-7417)
|
||||
* Fail to start if commit log replay detects a problem (CASSANDRA-7125)
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,24 +137,33 @@ 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);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
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));
|
||||
}
|
||||
|
|
@ -141,9 +184,23 @@ public class CqlTableTest extends PigTestBase
|
|||
@Test
|
||||
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();
|
||||
|
|
@ -170,9 +227,23 @@ public class CqlTableTest extends PigTestBase
|
|||
@Test
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue