private void SingleKeyTableTest(String initialQuery) throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { pig.setBatchOn(); pig.registerQuery(initialQuery); pig.registerQuery("insertformat= FOREACH moretestvalues GENERATE TOTUPLE(TOTUPLE('a',x)),TOTUPLE(y);"); pig.registerQuery("STORE insertformat INTO 'cql://cql3ks/test?" + defaultParameters + nativeParameters + "&output_query=UPDATE+cql3ks.test+set+b+%3D+%3F' USING CqlNativeStorage();"); pig.executeBatch(); //(5,5) //(6,6) //(4,4) //(2,2) //(3,3) //(1,1) //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"); while (it.hasNext()) { Tuple t = it.next(); Assert.assertEquals(t.get(0), t.get(1)); } }
@Test public void testCassandraStorageCounterCF() throws IOException, ClassNotFoundException, TException, TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException, IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException { pig.registerQuery("rows = LOAD 'cassandra://thriftKs/SomeApp?" + defaultParameters + "' USING CassandraStorage();"); //Test counter column family support pig.registerQuery("CC = load 'cassandra://thriftKs/CC?" + defaultParameters + "' using CassandraStorage();"); pig.registerQuery("total_hits = foreach CC generate key, SUM(columns.value);"); //(chuck,4) Iterator<Tuple> it = pig.openIterator("total_hits"); if (it.hasNext()) { Tuple t = it.next(); Assert.assertEquals(t.get(0), "chuck"); Assert.assertEquals(t.get(1), 4l); } }
@Test public void testCqlStorageSingleKeyTable() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { pig.setBatchOn(); pig.registerQuery("moretestvalues= LOAD 'cql://cql3ks/moredata?" + defaultParameters + "' USING CqlStorage();"); 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(); //(5,5) //(6,6) //(4,4) //(2,2) //(3,3) //(1,1) pig.registerQuery("result= LOAD 'cql://cql3ks/test?" + defaultParameters + "' USING CqlStorage();"); Iterator<Tuple> it = pig.openIterator("result"); if (it.hasNext()) { Tuple t = it.next(); Assert.assertEquals(t.get(0), t.get(1)); } }
@Test public void testCqlStorageCompositeKeyTable() throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, SchemaDisagreementException, IOException { pig.setBatchOn(); pig.registerQuery("moredata= LOAD 'cql://cql3ks/compmore?" + defaultParameters + "' USING CqlStorage();"); 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(); //(5,6,Fix,nomatch) //(3,3,Three,match) //(1,1,One,match) //(2,2,Two,match) //(7,7,Seven,match) //(8,8,Eight,match) //(6,5,Sive,nomatch) //(4,4,Four,match) //(9,10,Ninen,nomatch) pig.registerQuery("result= LOAD 'cql://cql3ks/compotable?" + defaultParameters + "' USING CqlStorage();"); Iterator<Tuple> it = pig.openIterator("result"); if (it.hasNext()) { Tuple t = it.next(); Assert.assertEquals(t.get(3), "match"); } }
private String getColumnValue(String ks, String cf, String colName, String key, String validator) throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, IOException { Cassandra.Client client = getClient(); client.set_keyspace(ks); ByteBuffer key_user_id = ByteBufferUtil.bytes(key); long timestamp = System.currentTimeMillis(); ColumnPath cp = new ColumnPath(cf); ColumnParent par = new ColumnParent(cf); cp.column = ByteBufferUtil.bytes(colName); // read ColumnOrSuperColumn got = client.get(key_user_id, cp, ConsistencyLevel.ONE); return parseType(validator).getString(got.getColumn().value); }
@BeforeClass public static void setup() throws TTransportException, IOException, InterruptedException, ConfigurationException, AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, CharacterCodingException, ClassNotFoundException, NoSuchFieldException, IllegalAccessException, InstantiationException { startCassandra(); setupDataByCql(statements); startHadoopCluster(); }
@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();"); }
@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();"); }
@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(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 + nativeParameters + "&output_query=UPDATE%20cql3ks.compotable%20SET%20d%20%3D%20%3F' USING CqlNativeStorage();"); pig.executeBatch(); //(5,6,Fix,nomatch) //(3,3,Three,match) //(1,1,One,match) //(2,2,Two,match) //(7,7,Seven,match) //(8,8,Eight,match) //(6,5,Sive,nomatch) //(4,4,Four,match) //(9,10,Ninen,nomatch) //input_cql=select * from compotable where token(a) > ? and token(a) <= ? pig.registerQuery("result= LOAD 'cql://cql3ks/compotable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20compotable%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()) { it.next(); count ++; } Assert.assertEquals(count, 9); }
@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=select%20*%20from%20collectiontable%20where%20token(m)%20%3E%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(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 + nativeParameters + "&output_query=update+cql3ks.collectiontable+set+n+%3D+%3F' USING CqlNativeStorage();"); pig.executeBatch(); //(book2,((m,mm),(n,nn))) //(book3,((m,mm),(n,nn))) //(book4,((m,mm),(n,nn))) //(book1,((m,mm),(n,nn))) //input_cql=select * from collectiontable where token(m) > ? and token(m) <= ? pig.registerQuery("result= LOAD 'cql://cql3ks/collectiontable?" + defaultParameters + nativeParameters + "&input_cql=select%20*%20from%20collectiontable%20where%20token(m)%20%3E%20%3F%20and%20token(m)%20%3C%3D%20%3F' USING CqlNativeStorage();"); Iterator<Tuple> it = pig.openIterator("result"); if (it.hasNext()) { Tuple t = it.next(); Tuple t1 = (Tuple) t.get(1); Assert.assertEquals(t1.size(), 2); Tuple element1 = (Tuple) t1.get(0); Tuple element2 = (Tuple) t1.get(1); Assert.assertEquals(element1.get(0), "m"); Assert.assertEquals(element1.get(1), "mm"); Assert.assertEquals(element2.get(0), "n"); Assert.assertEquals(element2.get(1), "nn"); } else { Assert.fail("Can't fetch any data"); } }
@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 CqlNativeStorage();"); //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 CqlNativeStorage();"); }
@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 CqlNativeStorage();"); }
@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 CqlNativeStorage();"); }
@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 CqlNativeStorage();"); }
@BeforeClass public static void setup() throws TTransportException, IOException, InterruptedException, ConfigurationException, AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, CharacterCodingException, ClassNotFoundException, NoSuchFieldException, IllegalAccessException, InstantiationException { startCassandra(); setupDataByCli(statements); startHadoopCluster(); }
@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();"); }
@Test public void testCassandraStorageFullCopy() throws IOException, ClassNotFoundException, TException, TimedOutException, NotFoundException, InvalidRequestException, NoSuchFieldException, UnavailableException, IllegalAccessException, InstantiationException, AuthenticationException, AuthorizationException { pig.setBatchOn(); pig.registerQuery("rows = LOAD 'cassandra://thriftKs/SomeApp?" + defaultParameters + "' USING CassandraStorage();"); //full copy pig.registerQuery("STORE rows INTO 'cassandra://thriftKs/CopyOfSomeApp?" + defaultParameters + "' USING CassandraStorage();"); pig.executeBatch(); Assert.assertEquals("User Qux", getColumnValue("thriftKs", "CopyOfSomeApp", "name", "qux", "UTF8Type")); Assert.assertEquals("dislike", getColumnValue("thriftKs", "CopyOfSomeApp", "vote_type", "qux", "UTF8Type")); Assert.assertEquals("64.7", getColumnValue("thriftKs", "CopyOfSomeApp", "percent", "qux", "FloatType")); }
private String getColumnValue(String ks, String cf, String colName, String key, String validator) throws AuthenticationException, AuthorizationException, InvalidRequestException, UnavailableException, TimedOutException, TException, NotFoundException, IOException { Cassandra.Client client = getClient(); client.set_keyspace(ks); ByteBuffer key_user_id = ByteBufferUtil.bytes(key); ColumnPath cp = new ColumnPath(cf); cp.column = ByteBufferUtil.bytes(colName); // read ColumnOrSuperColumn got = client.get(key_user_id, cp, ConsistencyLevel.ONE); return parseType(validator).getString(got.getColumn().value); }
