You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gora.apache.org by le...@apache.org on 2012/07/02 15:39:02 UTC
svn commit: r1356215 [1/3] - in /gora/branches/goraamazon: ./
gora-accumulo/src/main/java/org/apache/gora/accumulo/query/
gora-accumulo/src/main/java/org/apache/gora/accumulo/store/
gora-cassandra/src/main/java/org/apache/gora/cassandra/query/ gora-cas...
Author: lewismc
Date: Mon Jul 2 13:38:43 2012
New Revision: 1356215
URL: http://svn.apache.org/viewvc?rev=1356215&view=rev
Log:
Datastore for gora dynamodb phase 1
Added:
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/BeanFactoryWSImpl.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/PersistentWSBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/StateManagerWSImpl.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/ws/
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/ws/impl/
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/ws/impl/PartitionWSQueryImpl.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/ws/impl/QueryWSBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/ws/impl/ResultWSBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/WebServiceBackedDataStore.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/ws/
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/ws/impl/
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/ws/impl/WSBackedDataStoreBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/ws/impl/WSDataStoreBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/ws/impl/WSDataStoreFactory.java
gora/branches/goraamazon/gora-dynamodb/
gora/branches/goraamazon/gora-dynamodb/conf/
gora/branches/goraamazon/gora-dynamodb/conf/AwsCredentials.properties
gora/branches/goraamazon/gora-dynamodb/pom.xml
gora/branches/goraamazon/gora-dynamodb/src/
gora/branches/goraamazon/gora-dynamodb/src/main/
gora/branches/goraamazon/gora-dynamodb/src/main/java/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/query/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/query/DynamoDBColumn.java
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/query/DynamoDBQuery.java
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/query/DynamoDBResult.java
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/query/DynamoDBResultSet.java
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/query/DynamoDBRow.java
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/store/
gora/branches/goraamazon/gora-dynamodb/src/main/java/org/apache/gora/dynamodb/store/DynamoDBStore.java
Modified:
gora/branches/goraamazon/CHANGES.txt
gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloQuery.java
gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java
gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java
gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java
gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java
gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
gora/branches/goraamazon/gora-core/src/examples/java/org/apache/gora/examples/WebPageDataCreator.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumReader.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumWriter.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroQuery.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroResult.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/DataFileAvroResult.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/AvroStore.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/DataFileAvroStore.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordReader.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordWriter.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentDeserializer.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentNonReusingSerialization.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerialization.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerializer.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/memory/store/MemStore.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/BeanFactory.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/Persistent.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/StateManager.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/PersistentBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/StateManagerImpl.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/PartitionQuery.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Query.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Result.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/FileSplitPartitionQuery.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/PartitionQueryImpl.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/QueryBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/ResultBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/DataStore.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/DataStoreFactory.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/impl/DataStoreBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/store/impl/FileBackedDataStoreBase.java
gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/util/IOUtils.java
gora/branches/goraamazon/gora-core/src/test/java/org/apache/gora/avro/TestPersistentDatumReader.java
gora/branches/goraamazon/gora-core/src/test/java/org/apache/gora/avro/store/TestAvroStore.java
gora/branches/goraamazon/gora-core/src/test/java/org/apache/gora/mapreduce/MapReduceTestUtils.java
gora/branches/goraamazon/gora-core/src/test/java/org/apache/gora/persistency/impl/TestPersistentBase.java
gora/branches/goraamazon/gora-core/src/test/java/org/apache/gora/store/DataStoreTestBase.java
gora/branches/goraamazon/gora-core/src/test/java/org/apache/gora/store/DataStoreTestUtil.java
gora/branches/goraamazon/gora-hbase/src/main/java/org/apache/gora/hbase/query/HBaseGetResult.java
gora/branches/goraamazon/gora-hbase/src/main/java/org/apache/gora/hbase/query/HBaseQuery.java
gora/branches/goraamazon/gora-hbase/src/main/java/org/apache/gora/hbase/query/HBaseResult.java
gora/branches/goraamazon/gora-hbase/src/main/java/org/apache/gora/hbase/query/HBaseScannerResult.java
gora/branches/goraamazon/gora-hbase/src/main/java/org/apache/gora/hbase/store/HBaseStore.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/query/SqlQuery.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/query/SqlResult.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/statement/HSqlInsertUpdateStatement.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/statement/InsertUpdateStatement.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/statement/InsertUpdateStatementFactory.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/statement/MySqlInsertUpdateStatement.java
gora/branches/goraamazon/gora-sql/src/main/java/org/apache/gora/sql/store/SqlStore.java
gora/branches/goraamazon/gora-tutorial/src/main/java/org/apache/gora/tutorial/log/LogManager.java
gora/branches/goraamazon/pom.xml
Modified: gora/branches/goraamazon/CHANGES.txt
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/CHANGES.txt?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/CHANGES.txt (original)
+++ gora/branches/goraamazon/CHANGES.txt Mon Jul 2 13:38:43 2012
@@ -6,6 +6,8 @@ Gora Change Log
0.3 (trunk) Current Development:
+* GORA-103 Datastore for gora dynamodb - phase 1 (Renato Javier MarroquÃn Mogrovejo via lewismc)
+
* GORA-138 gora-cassandra array type support: Double fix for GORA-81 Replace CassandraStore#addOrUpdateField with TypeInferringSerializer to take advantage of when the value is already of type ByteBuffer. (Kazuomi Kashii via lewismc)
* GORA-139 Creates Cassandra column family with BytesType for column value validator (and comparators), instead of UTF8Type (Kazuomi Kashii via lewismc)
Modified: gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloQuery.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloQuery.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloQuery.java (original)
+++ gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloQuery.java Mon Jul 2 13:38:43 2012
@@ -16,14 +16,14 @@
*/
package org.apache.gora.accumulo.query;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.impl.QueryBase;
import org.apache.gora.store.DataStore;
/**
*
*/
-public class AccumuloQuery<K,T extends Persistent> extends QueryBase<K,T> {
+public class AccumuloQuery<K,T extends PersistentBase> extends QueryBase<K,T> {
public AccumuloQuery() {
super(null);
Modified: gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java (original)
+++ gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/query/AccumuloResult.java Mon Jul 2 13:38:43 2012
@@ -26,7 +26,7 @@ import org.apache.accumulo.core.data.Byt
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.gora.accumulo.store.AccumuloStore;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.impl.ResultBase;
import org.apache.gora.store.DataStore;
@@ -34,7 +34,7 @@ import org.apache.gora.store.DataStore;
/**
*
*/
-public class AccumuloResult<K,T extends Persistent> extends ResultBase<K,T> {
+public class AccumuloResult<K,T extends PersistentBase> extends ResultBase<K,T> {
private RowIterator iterator;
Modified: gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java (original)
+++ gora/branches/goraamazon/gora-accumulo/src/main/java/org/apache/gora/accumulo/store/AccumuloStore.java Mon Jul 2 13:38:43 2012
@@ -86,6 +86,7 @@ import org.apache.gora.persistency.State
import org.apache.gora.persistency.StateManager;
import org.apache.gora.persistency.StatefulHashMap;
import org.apache.gora.persistency.StatefulMap;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.PartitionQuery;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
@@ -101,7 +102,7 @@ import org.w3c.dom.NodeList;
/**
*
*/
-public class AccumuloStore<K,T extends Persistent> extends DataStoreBase<K,T> {
+public class AccumuloStore<K,T extends PersistentBase> extends DataStoreBase<K,T> {
protected static final String MOCK_PROPERTY = "accumulo.mock";
protected static final String INSTANCE_NAME_PROPERTY = "accumulo.instance";
Modified: gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java (original)
+++ gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraQuery.java Mon Jul 2 13:38:43 2012
@@ -22,11 +22,12 @@ import java.util.List;
import java.util.Map;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.impl.QueryBase;
import org.apache.gora.store.DataStore;
-public class CassandraQuery<K, T extends Persistent> extends QueryBase<K, T> {
+public class CassandraQuery<K, T extends PersistentBase> extends QueryBase<K, T> {
private Query<K, T> query;
Modified: gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java (original)
+++ gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/query/CassandraResult.java Mon Jul 2 13:38:43 2012
@@ -27,13 +27,14 @@ import me.prettyprint.cassandra.serializ
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.impl.ResultBase;
import org.apache.gora.store.DataStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class CassandraResult<K, T extends Persistent> extends ResultBase<K, T> {
+public class CassandraResult<K, T extends PersistentBase> extends ResultBase<K, T> {
public static final Logger LOG = LoggerFactory.getLogger(CassandraResult.class);
private int rowNumber;
Modified: gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java (original)
+++ gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraClient.java Mon Jul 2 13:38:43 2012
@@ -53,12 +53,13 @@ import org.apache.avro.util.Utf8;
import org.apache.gora.cassandra.query.CassandraQuery;
import org.apache.gora.mapreduce.GoraRecordReader;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.util.ByteUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class CassandraClient<K, T extends Persistent> {
+public class CassandraClient<K, T extends PersistentBase> {
public static final Logger LOG = LoggerFactory.getLogger(CassandraClient.class);
private Cluster cluster;
Modified: gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java (original)
+++ gora/branches/goraamazon/gora-cassandra/src/main/java/org/apache/gora/cassandra/store/CassandraStore.java Mon Jul 2 13:38:43 2012
@@ -61,7 +61,7 @@ import org.apache.gora.store.impl.DataSt
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class CassandraStore<K, T extends Persistent> extends DataStoreBase<K, T> {
+public class CassandraStore<K, T extends PersistentBase> extends DataStoreBase<K, T> {
public static final Logger LOG = LoggerFactory.getLogger(CassandraStore.class);
private CassandraClient<K, T> cassandraClient = new CassandraClient<K, T>();
@@ -236,7 +236,7 @@ public class CassandraStore<K, T extends
}
@Override
- public T get(K key, String[] fields) throws IOException {
+ public T get(K key, String[] fields) throws IOException, Exception {
CassandraQuery<K,T> query = new CassandraQuery<K,T>();
query.setDataStore(this);
query.setKeyRange(key, key);
@@ -285,8 +285,8 @@ public class CassandraStore<K, T extends
Type type = fieldSchema.getType();
switch(type) {
case RECORD:
- Persistent persistent = (Persistent) fieldValue;
- Persistent newRecord = persistent.newInstance(new StateManagerImpl());
+ PersistentBase persistent = (PersistentBase) fieldValue;
+ PersistentBase newRecord = (PersistentBase) persistent.newInstance(new StateManagerImpl());
for (Field member: fieldSchema.getFields()) {
newRecord.put(member.pos(), persistent.get(member.pos()));
}
Modified: gora/branches/goraamazon/gora-core/src/examples/java/org/apache/gora/examples/WebPageDataCreator.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/examples/java/org/apache/gora/examples/WebPageDataCreator.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/examples/java/org/apache/gora/examples/WebPageDataCreator.java (original)
+++ gora/branches/goraamazon/gora-core/src/examples/java/org/apache/gora/examples/WebPageDataCreator.java Mon Jul 2 13:38:43 2012
@@ -109,30 +109,35 @@ public class WebPageDataCreator {
public static void createWebPageData(DataStore<String, WebPage> dataStore)
throws IOException {
- WebPage page;
- log.info("creating web page data");
-
- for(int i=0; i<URLS.length; i++) {
- page = new WebPage();
- page.setUrl(new Utf8(URLS[i]));
- page.setContent(ByteBuffer.wrap(CONTENTS[i].getBytes()));
- for(String token : CONTENTS[i].split(" ")) {
- page.addToParsedContent(new Utf8(token));
- }
-
- for(int j=0; j<LINKS[i].length; j++) {
- page.putToOutlinks(new Utf8(URLS[LINKS[i][j]]), new Utf8(ANCHORS[i][j]));
- }
-
- Metadata metadata = new Metadata();
- metadata.setVersion(1);
- metadata.putToData(new Utf8("metakey"), new Utf8("metavalue"));
- page.setMetadata(metadata);
-
- dataStore.put(URLS[i], page);
- }
- dataStore.flush();
- log.info("finished creating web page data");
+ try{
+ WebPage page;
+ log.info("creating web page data");
+
+ for(int i=0; i<URLS.length; i++) {
+ page = new WebPage();
+ page.setUrl(new Utf8(URLS[i]));
+ page.setContent(ByteBuffer.wrap(CONTENTS[i].getBytes()));
+ for(String token : CONTENTS[i].split(" ")) {
+ page.addToParsedContent(new Utf8(token));
+ }
+
+ for(int j=0; j<LINKS[i].length; j++) {
+ page.putToOutlinks(new Utf8(URLS[LINKS[i][j]]), new Utf8(ANCHORS[i][j]));
+ }
+
+ Metadata metadata = new Metadata();
+ metadata.setVersion(1);
+ metadata.putToData(new Utf8("metakey"), new Utf8("metavalue"));
+ page.setMetadata(metadata);
+
+ dataStore.put(URLS[i], page);
+ }
+ dataStore.flush();
+ log.info("finished creating web page data");
+ }
+ catch(Exception e){
+ log.info("error creating web page data");
+ }
}
public int run(String[] args) throws Exception {
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumReader.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumReader.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumReader.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumReader.java Mon Jul 2 13:38:43 2012
@@ -38,13 +38,14 @@ import org.apache.gora.persistency.Persi
import org.apache.gora.persistency.State;
import org.apache.gora.persistency.StatefulHashMap;
import org.apache.gora.persistency.StatefulMap;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.persistency.impl.StateManagerImpl;
import org.apache.gora.util.IOUtils;
/**
* PersistentDatumReader reads, fields' dirty and readable information.
*/
-public class PersistentDatumReader<T extends Persistent>
+public class PersistentDatumReader<T extends PersistentBase>
extends SpecificDatumReader<T> {
private Schema rootSchema;
@@ -212,7 +213,7 @@ public class PersistentDatumReader<T ext
}
public Persistent clone(Persistent persistent, Schema schema) {
- Persistent cloned = persistent.newInstance(new StateManagerImpl());
+ Persistent cloned = (PersistentBase)persistent.newInstance(new StateManagerImpl());
List<Field> fields = schema.getFields();
for(Field field: fields) {
int pos = field.pos();
@@ -220,10 +221,10 @@ public class PersistentDatumReader<T ext
case MAP :
case ARRAY :
case RECORD :
- case STRING : cloned.put(pos, cloneObject(
- field.schema(), persistent.get(pos), cloned.get(pos))); break;
+ case STRING : ((PersistentBase)cloned).put(pos, cloneObject(
+ field.schema(), ((PersistentBase)persistent).get(pos), ((PersistentBase)cloned).get(pos))); break;
case NULL : break;
- default : cloned.put(pos, persistent.get(pos)); break;
+ default : ((PersistentBase)cloned).put(pos, ((PersistentBase)persistent).get(pos)); break;
}
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumWriter.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumWriter.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumWriter.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/PersistentDatumWriter.java Mon Jul 2 13:38:43 2012
@@ -26,16 +26,16 @@ import org.apache.avro.Schema.Field;
import org.apache.avro.io.Encoder;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.avro.util.Utf8;
-import org.apache.gora.persistency.Persistent;
import org.apache.gora.persistency.State;
import org.apache.gora.persistency.StateManager;
import org.apache.gora.persistency.StatefulMap;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.util.IOUtils;
/**
* PersistentDatumWriter writes, fields' dirty and readable information.
*/
-public class PersistentDatumWriter<T extends Persistent>
+public class PersistentDatumWriter<T extends PersistentBase>
extends SpecificDatumWriter<T> {
private T persistent = null;
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroQuery.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroQuery.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroQuery.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroQuery.java Mon Jul 2 13:38:43 2012
@@ -18,8 +18,11 @@
package org.apache.gora.avro.query;
+import java.io.IOException;
+
import org.apache.gora.avro.store.AvroStore;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
+import org.apache.gora.query.Result;
import org.apache.gora.query.impl.QueryBase;
/**
@@ -27,7 +30,7 @@ import org.apache.gora.query.impl.QueryB
* most of the operations for Query, like setting start,end keys is not
* supported. Setting query limit is supported.
*/
-public class AvroQuery<K, T extends Persistent> extends QueryBase<K,T> {
+public class AvroQuery<K, T extends PersistentBase> extends QueryBase<K,T> {
public AvroQuery() {
super(null);
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroResult.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroResult.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroResult.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/AvroResult.java Mon Jul 2 13:38:43 2012
@@ -25,13 +25,13 @@ import org.apache.avro.AvroTypeException
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.gora.avro.store.AvroStore;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.impl.ResultBase;
/**
* Adapter to convert DatumReader to Result.
*/
-public class AvroResult<K, T extends Persistent> extends ResultBase<K, T> {
+public class AvroResult<K, T extends PersistentBase> extends ResultBase<K, T> {
private DatumReader<T> reader;
private Decoder decoder;
@@ -43,7 +43,6 @@ public class AvroResult<K, T extends Per
this.decoder = decoder;
}
- @Override
public void close() throws IOException {
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/DataFileAvroResult.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/DataFileAvroResult.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/DataFileAvroResult.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/query/DataFileAvroResult.java Mon Jul 2 13:38:43 2012
@@ -23,6 +23,7 @@ import java.io.IOException;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.SeekableInput;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.impl.ResultBase;
import org.apache.gora.store.DataStore;
@@ -30,7 +31,7 @@ import org.apache.gora.store.DataStore;
/**
* An Avro {@link DataFileReader} backed Result.
*/
-public class DataFileAvroResult<K, T extends Persistent> extends ResultBase<K, T> {
+public class DataFileAvroResult<K, T extends PersistentBase> extends ResultBase<K, T> {
private SeekableInput in;
private DataFileReader<T> reader;
@@ -58,9 +59,9 @@ public class DataFileAvroResult<K, T ext
@Override
public void close() throws IOException {
- if(reader != null)
- reader.close();
- reader = null;
+ if(reader != null)
+ reader.close();
+ reader = null;
}
@Override
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/AvroStore.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/AvroStore.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/AvroStore.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/AvroStore.java Mon Jul 2 13:38:43 2012
@@ -35,7 +35,7 @@ import org.apache.avro.specific.Specific
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.gora.avro.query.AvroQuery;
import org.apache.gora.avro.query.AvroResult;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
import org.apache.gora.query.impl.FileSplitPartitionQuery;
@@ -50,7 +50,7 @@ import org.apache.hadoop.conf.Configurat
* AvroDataStore supports Binary and JSON serializations.
* @param <T>
*/
-public class AvroStore<K, T extends Persistent>
+public class AvroStore<K, T extends PersistentBase>
extends FileBackedDataStoreBase<K, T> implements Configurable {
/** The property key specifying avro encoder/decoder type to use. Can take values
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/DataFileAvroStore.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/DataFileAvroStore.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/DataFileAvroStore.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/avro/store/DataFileAvroStore.java Mon Jul 2 13:38:43 2012
@@ -25,6 +25,7 @@ import org.apache.avro.file.DataFileWrit
import org.apache.gora.avro.mapreduce.FsInput;
import org.apache.gora.avro.query.DataFileAvroResult;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
import org.apache.gora.query.impl.FileSplitPartitionQuery;
@@ -36,7 +37,7 @@ import org.apache.hadoop.fs.Path;
* DataFile{Writer,Reader}'s as a backend. This datastore supports
* mapreduce.
*/
-public class DataFileAvroStore<K, T extends Persistent> extends AvroStore<K, T> {
+public class DataFileAvroStore<K, T extends PersistentBase> extends AvroStore<K, T> {
public DataFileAvroStore() {
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraInputFormat.java Mon Jul 2 13:38:43 2012
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.PartitionQuery;
import org.apache.gora.query.Query;
import org.apache.gora.query.impl.FileSplitPartitionQuery;
@@ -52,7 +53,7 @@ import org.apache.hadoop.mapreduce.lib.i
*
* @see GoraMapper
*/
-public class GoraInputFormat<K, T extends Persistent>
+public class GoraInputFormat<K, T extends PersistentBase>
extends InputFormat<K, T> implements Configurable {
public static final String QUERY_KEY = "gora.inputformat.query";
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordReader.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordReader.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordReader.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordReader.java Mon Jul 2 13:38:43 2012
@@ -21,8 +21,10 @@ package org.apache.gora.mapreduce;
import java.io.IOException;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
+import org.apache.gora.query.impl.ResultBase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
@@ -33,7 +35,7 @@ import org.slf4j.LoggerFactory;
/**
* An adapter for Result to Hadoop RecordReader.
*/
-public class GoraRecordReader<K, T extends Persistent> extends RecordReader<K,T> {
+public class GoraRecordReader<K, T extends PersistentBase> extends RecordReader<K,T> {
public static final Logger LOG = LoggerFactory.getLogger(GoraRecordReader.class);
public static final String BUFFER_LIMIT_READ_NAME = "gora.buffer.read.limit";
@@ -62,7 +64,7 @@ public class GoraRecordReader<K, T exten
this.query.setLimit(recordsMax);
}
- public void executeQuery() throws IOException {
+ public void executeQuery() throws IOException, Exception {
this.result = query.execute();
}
@@ -78,7 +80,12 @@ public class GoraRecordReader<K, T exten
@Override
public float getProgress() throws IOException, InterruptedException {
- return result.getProgress();
+ try{
+ return result.getProgress();
+ }
+ catch(Exception e){
+ return 0;
+ }
}
@Override
@@ -87,31 +94,36 @@ public class GoraRecordReader<K, T exten
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
- if (counter.isModulo()) {
- boolean firstBatch = (this.result == null);
- if (! firstBatch) {
- this.query.setStartKey(this.result.getKey());
- if (this.query.getLimit() == counter.getRecordsMax()) {
- this.query.setLimit(counter.getRecordsMax() + 1);
- }
- }
- if (this.result != null) {
- this.result.close();
- }
-
- executeQuery();
-
- if (! firstBatch) {
- // skip first result
- this.result.next();
- }
- }
-
- counter.increment();
- return this.result.next();
+ try{
+ if (counter.isModulo()) {
+ boolean firstBatch = (this.result == null);
+ if (! firstBatch) {
+ this.query.setStartKey(this.result.getKey());
+ if (this.query.getLimit() == counter.getRecordsMax()) {
+ this.query.setLimit(counter.getRecordsMax() + 1);
+ }
+ }
+ if (this.result != null) {
+ this.result.close();
+ }
+
+ executeQuery();
+
+ if (! firstBatch) {
+ // skip first result
+ this.result.next();
+ }
+ }
+
+ counter.increment();
+ return this.result.next();
+ }
+ catch(Exception e){
+ return false;
+ }
}
- @Override
+ //@Override
public void close() throws IOException {
if (result != null) {
result.close();
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordWriter.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordWriter.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordWriter.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/GoraRecordWriter.java Mon Jul 2 13:38:43 2012
@@ -52,17 +52,25 @@ public class GoraRecordWriter<K, T> exte
@Override
public void close(TaskAttemptContext context) throws IOException,
InterruptedException {
- store.close();
+ try{
+ store.close();
+ }catch(Exception e){
+ LOG.info("Exception at GoraRecordWriter.class while closing datastore." + e.getMessage());
+ }
}
@Override
public void write(K key, T value) throws IOException, InterruptedException {
- store.put(key, (Persistent) value);
-
- counter.increment();
- if (counter.isModulo()) {
- LOG.info("Flushing the datastore after " + counter.getRecordsNumber() + " records");
- store.flush();
- }
+ try{
+ store.put(key, (Persistent) value);
+
+ counter.increment();
+ if (counter.isModulo()) {
+ LOG.info("Flushing the datastore after " + counter.getRecordsNumber() + " records");
+ store.flush();
+ }
+ }catch(Exception e){
+ LOG.info("Exception at GoraRecordWriter.class while writing to datastore." + e.getMessage());
+ }
}
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentDeserializer.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentDeserializer.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentDeserializer.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentDeserializer.java Mon Jul 2 13:38:43 2012
@@ -25,6 +25,7 @@ import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.gora.avro.PersistentDatumReader;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.util.AvroUtils;
import org.apache.hadoop.io.serializer.Deserializer;
@@ -33,26 +34,26 @@ import org.apache.hadoop.io.serializer.D
* with {@link BinaryDecoder}.
*/
public class PersistentDeserializer
- implements Deserializer<Persistent> {
+ implements Deserializer<PersistentBase> {
private BinaryDecoder decoder;
- private Class<? extends Persistent> persistentClass;
+ private Class<? extends PersistentBase> persistentClass;
private boolean reuseObjects;
- private PersistentDatumReader<Persistent> datumReader;
+ private PersistentDatumReader<PersistentBase> datumReader;
- public PersistentDeserializer(Class<? extends Persistent> c, boolean reuseObjects) {
+ public PersistentDeserializer(Class<? extends PersistentBase> c, boolean reuseObjects) {
this.persistentClass = c;
this.reuseObjects = reuseObjects;
try {
Schema schema = AvroUtils.getSchema(persistentClass);
- datumReader = new PersistentDatumReader<Persistent>(schema, true);
+ datumReader = new PersistentDatumReader<PersistentBase>(schema, true);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
- @Override
+ //@Override
public void open(InputStream in) throws IOException {
/* It is very important to use a direct buffer, since Hadoop
* supplies an input stream that is only valid until the end of one
@@ -64,11 +65,11 @@ public class PersistentDeserializer
.createBinaryDecoder(in, decoder);
}
- @Override
+ //@Override
public void close() throws IOException { }
@Override
- public Persistent deserialize(Persistent persistent) throws IOException {
+ public PersistentBase deserialize(PersistentBase persistent) throws IOException {
return datumReader.read(reuseObjects ? persistent : null, decoder);
}
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentNonReusingSerialization.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentNonReusingSerialization.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentNonReusingSerialization.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentNonReusingSerialization.java Mon Jul 2 13:38:43 2012
@@ -18,12 +18,13 @@
package org.apache.gora.mapreduce;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.hadoop.io.serializer.Deserializer;
import org.apache.hadoop.io.serializer.Serialization;
import org.apache.hadoop.io.serializer.Serializer;
public class PersistentNonReusingSerialization
-implements Serialization<Persistent> {
+implements Serialization<PersistentBase> {
@Override
public boolean accept(Class<?> c) {
@@ -31,12 +32,12 @@ implements Serialization<Persistent> {
}
@Override
- public Deserializer<Persistent> getDeserializer(Class<Persistent> c) {
+ public Deserializer<PersistentBase> getDeserializer(Class<PersistentBase> c) {
return new PersistentDeserializer(c, false);
}
@Override
- public Serializer<Persistent> getSerializer(Class<Persistent> c) {
+ public Serializer<PersistentBase> getSerializer(Class<PersistentBase> c) {
return new PersistentSerializer();
}
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerialization.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerialization.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerialization.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerialization.java Mon Jul 2 13:38:43 2012
@@ -17,13 +17,14 @@
*/
package org.apache.gora.mapreduce;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.persistency.Persistent;
import org.apache.hadoop.io.serializer.Deserializer;
import org.apache.hadoop.io.serializer.Serialization;
import org.apache.hadoop.io.serializer.Serializer;
public class PersistentSerialization
-implements Serialization<Persistent> {
+implements Serialization<PersistentBase> {
@Override
public boolean accept(Class<?> c) {
@@ -31,12 +32,12 @@ implements Serialization<Persistent> {
}
@Override
- public Deserializer<Persistent> getDeserializer(Class<Persistent> c) {
+ public Deserializer<PersistentBase> getDeserializer(Class<PersistentBase> c) {
return new PersistentDeserializer(c, true);
}
@Override
- public Serializer<Persistent> getSerializer(Class<Persistent> c) {
+ public Serializer<PersistentBase> getSerializer(Class<PersistentBase> c) {
return new PersistentSerializer();
}
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerializer.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerializer.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerializer.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/mapreduce/PersistentSerializer.java Mon Jul 2 13:38:43 2012
@@ -22,34 +22,34 @@ import java.io.OutputStream;
import org.apache.avro.io.BinaryEncoder;
import org.apache.gora.avro.PersistentDatumWriter;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.hadoop.io.serializer.Serializer;
/**
* Hadoop serializer using {@link PersistentDatumWriter}
* with {@link BinaryEncoder}.
*/
-public class PersistentSerializer implements Serializer<Persistent> {
+public class PersistentSerializer implements Serializer<PersistentBase> {
- private PersistentDatumWriter<Persistent> datumWriter;
+ private PersistentDatumWriter<PersistentBase> datumWriter;
private BinaryEncoder encoder;
public PersistentSerializer() {
- this.datumWriter = new PersistentDatumWriter<Persistent>();
+ this.datumWriter = new PersistentDatumWriter<PersistentBase>();
}
- @Override
+ //@Override
public void close() throws IOException {
encoder.flush();
}
- @Override
+ //@Override
public void open(OutputStream out) throws IOException {
encoder = new BinaryEncoder(out);
}
@Override
- public void serialize(Persistent persistent) throws IOException {
+ public void serialize(PersistentBase persistent) throws IOException {
datumWriter.setSchema(persistent.getSchema());
datumWriter.setPersistent(persistent);
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/memory/store/MemStore.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/memory/store/MemStore.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/memory/store/MemStore.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/memory/store/MemStore.java Mon Jul 2 13:38:43 2012
@@ -27,6 +27,7 @@ import java.util.NavigableMap;
import java.util.TreeMap;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.persistency.impl.StateManagerImpl;
import org.apache.gora.query.PartitionQuery;
import org.apache.gora.query.Query;
@@ -40,9 +41,9 @@ import org.apache.gora.store.impl.DataSt
/**
* Memory based {@link DataStore} implementation for tests.
*/
-public class MemStore<K, T extends Persistent> extends DataStoreBase<K, T> {
+public class MemStore<K, T extends PersistentBase> extends DataStoreBase<K, T> {
- public static class MemQuery<K, T extends Persistent> extends QueryBase<K, T> {
+ public static class MemQuery<K, T extends PersistentBase> extends QueryBase<K, T> {
public MemQuery() {
super(null);
}
@@ -51,7 +52,7 @@ public class MemStore<K, T extends Persi
}
}
- public static class MemResult<K, T extends Persistent> extends ResultBase<K, T> {
+ public static class MemResult<K, T extends PersistentBase> extends ResultBase<K, T> {
private NavigableMap<K, T> map;
private Iterator<K> iterator;
public MemResult(DataStore<K, T> dataStore, Query<K, T> query
@@ -60,7 +61,7 @@ public class MemStore<K, T extends Persi
this.map = map;
iterator = map.navigableKeySet().iterator();
}
- @Override
+ //@Override
public void close() throws IOException { }
@Override
public float getProgress() throws IOException {
@@ -97,15 +98,19 @@ public class MemStore<K, T extends Persi
@Override
public long deleteByQuery(Query<K, T> query) throws IOException {
- long deletedRows = 0;
- Result<K,T> result = query.execute();
-
- while(result.next()) {
- if(delete(result.getKey()))
- deletedRows++;
- }
-
- return 0;
+ try{
+ long deletedRows = 0;
+ Result<K,T> result = query.execute();
+
+ while(result.next()) {
+ if(delete(result.getKey()))
+ deletedRows++;
+ }
+ return 0;
+ }
+ catch(Exception e){
+ return 0;
+ }
}
@Override
@@ -144,7 +149,7 @@ public class MemStore<K, T extends Persi
T newObj = (T) obj.newInstance(new StateManagerImpl());
for(String field:fields) {
int index = newObj.getFieldIndex(field);
- newObj.put(index, obj.get(index));
+ ((PersistentBase)newObj).put(index, ((PersistentBase)obj).get(index));
}
return newObj;
}
@@ -163,8 +168,7 @@ public class MemStore<K, T extends Persi
/**
* Returns a single partition containing the original query
*/
- public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query)
- throws IOException {
+ public List<PartitionQuery<K, T>> getPartitions(Query<K, T> query){
List<PartitionQuery<K, T>> list = new ArrayList<PartitionQuery<K,T>>();
list.add(new PartitionQueryImpl<K, T>(query));
return list;
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/BeanFactory.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/BeanFactory.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/BeanFactory.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/BeanFactory.java Mon Jul 2 13:38:43 2012
@@ -21,7 +21,7 @@ package org.apache.gora.persistency;
/**
* BeanFactory's enable contruction of keys and Persistent objects.
*/
-public interface BeanFactory<K, T extends Persistent> {
+public interface BeanFactory<K, T>{
/**
* Constructs a new instance of the key class
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/Persistent.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/Persistent.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/Persistent.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/Persistent.java Mon Jul 2 13:38:43 2012
@@ -17,12 +17,10 @@
*/
package org.apache.gora.persistency;
-import org.apache.avro.specific.SpecificRecord;
-
/**
* Objects that are persisted by Gora implements this interface.
*/
-public interface Persistent extends SpecificRecord, Cloneable {
+public interface Persistent extends Cloneable{
/**
* Returns the StateManager which manages the persistent
@@ -183,4 +181,5 @@ public interface Persistent extends Spec
void clearReadable();
Persistent clone();
+
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/StateManager.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/StateManager.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/StateManager.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/StateManager.java Mon Jul 2 13:38:43 2012
@@ -21,7 +21,7 @@ package org.apache.gora.persistency;
/**
* StateManager manages objects state for persistency.
*/
-public interface StateManager {
+public interface StateManager{
/**
* If one state manager is allocated per persistent object,
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/PersistentBase.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/PersistentBase.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/PersistentBase.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/PersistentBase.java Mon Jul 2 13:38:43 2012
@@ -35,7 +35,7 @@ import org.apache.gora.persistency.State
* Base classs implementing common functionality for Persistent
* classes.
*/
-public abstract class PersistentBase implements Persistent {
+public abstract class PersistentBase implements Persistent, SpecificRecord {
protected static Map<Class<?>, Map<String, Integer>> FIELD_MAP =
new HashMap<Class<?>, Map<String,Integer>>();
@@ -43,8 +43,8 @@ public abstract class PersistentBase imp
protected static Map<Class<?>, String[]> FIELDS =
new HashMap<Class<?>, String[]>();
- protected static final PersistentDatumReader<Persistent> datumReader =
- new PersistentDatumReader<Persistent>();
+ protected static final PersistentDatumReader<PersistentBase> datumReader =
+ new PersistentDatumReader<PersistentBase>();
private StateManager stateManager;
@@ -230,7 +230,7 @@ public abstract class PersistentBase imp
clearReadable(getFieldIndex(field));
}
- @Override
+ //@Override
public boolean equals(Object o) {
if (this == o) return true;
if (!(o instanceof SpecificRecord)) return false;
@@ -241,7 +241,7 @@ public abstract class PersistentBase imp
return this.hashCode() == r2.hashCode();
}
- @Override
+ //@Override
public int hashCode() {
final int prime = 31;
int result = 1;
@@ -280,7 +280,7 @@ public abstract class PersistentBase imp
return datumReader.clone(this, getSchema());
}
- @Override
+ //@Override
public String toString() {
StringBuilder builder = new StringBuilder();
builder.append(super.toString());
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/StateManagerImpl.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/StateManagerImpl.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/StateManagerImpl.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/impl/StateManagerImpl.java Mon Jul 2 13:38:43 2012
@@ -38,8 +38,8 @@ public class StateManagerImpl implements
}
public void setManagedPersistent(Persistent persistent) {
- dirtyBits = new BitSet(persistent.getSchema().getFields().size());
- readableBits = new BitSet(persistent.getSchema().getFields().size());
+ dirtyBits = new BitSet(((PersistentBase)persistent).getSchema().getFields().size());
+ readableBits = new BitSet(((PersistentBase)persistent).getSchema().getFields().size());
isNew = true;
}
Added: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/BeanFactoryWSImpl.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/BeanFactoryWSImpl.java?rev=1356215&view=auto
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/BeanFactoryWSImpl.java (added)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/BeanFactoryWSImpl.java Mon Jul 2 13:38:43 2012
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gora.persistency.ws.impl;
+
+import java.lang.reflect.Constructor;
+
+import org.apache.gora.persistency.BeanFactory;
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.util.ReflectionUtils;
+
+/**
+ * A default implementation of the {@link BeanFactory} interface. Constructs
+ * the keys using by reflection, {@link Persistent} objects by calling
+ * {@link Persistent#newInstance(org.apache.gora.persistency.StateManager)}.
+ */
+public class BeanFactoryWSImpl<K, T extends Persistent> implements BeanFactory<K, T> {
+
+ private Class<K> keyClass;
+ private Class<T> persistentClass;
+
+ private Constructor<K> keyConstructor;
+
+ private K key;
+ private T persistent;
+
+ private boolean isKeyPersistent = false;
+
+ public BeanFactoryWSImpl(Class<K> keyClass, Class<T> persistentClass) {
+ this.keyClass = keyClass;
+ this.persistentClass = persistentClass;
+
+ try {
+ if(ReflectionUtils.hasConstructor(keyClass)) {
+ this.keyConstructor = ReflectionUtils.getConstructor(keyClass);
+ this.key = keyConstructor.newInstance(ReflectionUtils.EMPTY_OBJECT_ARRAY);
+ }
+ this.persistent = ReflectionUtils.newInstance(persistentClass);
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+
+ isKeyPersistent = Persistent.class.isAssignableFrom(keyClass);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public K newKey() throws Exception {
+ if(isKeyPersistent)
+ return (K)((Persistent)key).newInstance(new StateManagerWSImpl());
+ else if(keyConstructor == null) {
+ throw new RuntimeException("Key class does not have a no-arg constructor");
+ }
+ else
+ return keyConstructor.newInstance(ReflectionUtils.EMPTY_OBJECT_ARRAY);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public T newPersistent() {
+ return (T) persistent.newInstance(new StateManagerWSImpl());
+ }
+
+ @Override
+ public K getCachedKey() {
+ return key;
+ }
+
+ @Override
+ public T getCachedPersistent() {
+ return persistent;
+ }
+
+ @Override
+ public Class<K> getKeyClass() {
+ return keyClass;
+ }
+
+ @Override
+ public Class<T> getPersistentClass() {
+ return persistentClass;
+ }
+
+ public boolean isKeyPersistent() {
+ return isKeyPersistent;
+ }
+}
Added: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/PersistentWSBase.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/PersistentWSBase.java?rev=1356215&view=auto
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/PersistentWSBase.java (added)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/PersistentWSBase.java Mon Jul 2 13:38:43 2012
@@ -0,0 +1,298 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gora.persistency.ws.impl;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.StateManager;
+
+/**
+ * Base classs implementing common functionality for Persistent
+ * classes.
+ */
+public abstract class PersistentWSBase implements Persistent {
+
+ protected static Map<Class<?>, Map<String, Integer>> FIELD_MAP =
+ new HashMap<Class<?>, Map<String,Integer>>();
+
+ protected static Map<Class<?>, String[]> FIELDS =
+ new HashMap<Class<?>, String[]>();
+
+ private StateManager stateManager;
+
+ protected PersistentWSBase() {
+ this(new StateManagerWSImpl());
+ }
+
+ protected PersistentWSBase(StateManager stateManager) {
+ this.stateManager = stateManager;
+ stateManager.setManagedPersistent(this);
+ }
+
+ /** Subclasses should call this function for all the persistable fields
+ * in the class to register them.
+ * @param clazz the Persistent class
+ * @param fields the name of the fields of the class
+ */
+ protected static void registerFields(Class<?> clazz, String... fields) {
+ FIELDS.put(clazz, fields);
+ int fieldsLength = fields == null ? 0 :fields.length;
+ HashMap<String, Integer> map = new HashMap<String, Integer>(fieldsLength);
+
+ for(int i=0; i < fieldsLength; i++) {
+ map.put(fields[i], i);
+ }
+ FIELD_MAP.put(clazz, map);
+ }
+
+ @Override
+ public StateManager getStateManager() {
+ return stateManager;
+ }
+
+ @Override
+ public String[] getFields() {
+ return FIELDS.get(getClass());
+ }
+
+ @Override
+ public String getField(int index) {
+ return FIELDS.get(getClass())[index];
+ }
+
+ @Override
+ public int getFieldIndex(String field) {
+ return FIELD_MAP.get(getClass()).get(field);
+ }
+
+ @Override
+ public void clear() {
+ /*List<Field> fields = getSchema().getFields();
+
+ for(int i=0; i<getFields().length; i++) {
+ switch(fields.get(i).schema().getType()) {
+ case MAP:
+ if(get(i) != null) {
+ if (get(i) instanceof StatefulHashMap) {
+ ((StatefulHashMap)get(i)).reuse();
+ } else {
+ ((Map)get(i)).clear();
+ }
+ }
+ break;
+ case ARRAY:
+ if(get(i) != null) {
+ if(get(i) instanceof ListGenericArray) {
+ ((ListGenericArray)get(i)).clear();
+ } else {
+ put(i, new ListGenericArray(fields.get(i).schema()));
+ }
+ }
+ break;
+ case RECORD :
+ Persistent field = ((Persistent)get(i));
+ if(field != null) field.clear();
+ break;
+ case BOOLEAN: put(i, false); break;
+ case INT : put(i, 0); break;
+ case DOUBLE : put(i, 0d); break;
+ case FLOAT : put(i, 0f); break;
+ case LONG : put(i, 0l); break;
+ case NULL : break;
+ default : put(i, null); break;
+ }
+ }*/
+ clearDirty();
+ clearReadable();
+ }
+
+ @Override
+ public boolean isNew() {
+ return getStateManager().isNew(this);
+ }
+
+ @Override
+ public void setNew() {
+ getStateManager().setNew(this);
+ }
+
+ @Override
+ public void clearNew() {
+ getStateManager().clearNew(this);
+ }
+
+ @Override
+ public boolean isDirty() {
+ return getStateManager().isDirty(this);
+ }
+
+ @Override
+ public boolean isDirty(int fieldIndex) {
+ return getStateManager().isDirty(this, fieldIndex);
+ }
+
+ @Override
+ public boolean isDirty(String field) {
+ return isDirty(getFieldIndex(field));
+ }
+
+ @Override
+ public void setDirty() {
+ getStateManager().setDirty(this);
+ }
+
+ @Override
+ public void setDirty(int fieldIndex) {
+ getStateManager().setDirty(this, fieldIndex);
+ }
+
+ @Override
+ public void setDirty(String field) {
+ setDirty(getFieldIndex(field));
+ }
+
+ @Override
+ public void clearDirty(int fieldIndex) {
+ getStateManager().clearDirty(this, fieldIndex);
+ }
+
+ @Override
+ public void clearDirty(String field) {
+ clearDirty(getFieldIndex(field));
+ }
+
+ @Override
+ public void clearDirty() {
+ getStateManager().clearDirty(this);
+ }
+
+ @Override
+ public boolean isReadable(int fieldIndex) {
+ return getStateManager().isReadable(this, fieldIndex);
+ }
+
+ @Override
+ public boolean isReadable(String field) {
+ return isReadable(getFieldIndex(field));
+ }
+
+ @Override
+ public void setReadable(int fieldIndex) {
+ getStateManager().setReadable(this, fieldIndex);
+ }
+
+ @Override
+ public void setReadable(String field) {
+ setReadable(getFieldIndex(field));
+ }
+
+ @Override
+ public void clearReadable() {
+ getStateManager().clearReadable(this);
+ }
+
+ @Override
+ public void clearReadable(int fieldIndex) {
+ getStateManager().clearReadable(this, fieldIndex);
+ }
+
+ @Override
+ public void clearReadable(String field) {
+ clearReadable(getFieldIndex(field));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ /*if (!(o instanceof SpecificRecord)) return false;
+
+ SpecificRecord r2 = (SpecificRecord)o;
+ if (!this.getSchema().equals(r2.getSchema())) return false;
+
+ return this.hashCode() == r2.hashCode();*/
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = 1;
+ /* List<Field> fields = this.getSchema().getFields();
+ int end = fields.size();
+ for (int i = 0; i < end; i++) {
+ result = prime * result + getFieldHashCode(i, fields.get(i));
+ }
+ */
+ return result;
+ }
+
+ /*private int getFieldHashCode(int i, Field field) {
+ Object o = get(i);
+ if(o == null)
+ return 0;
+
+ if(field.schema().getType() == Type.BYTES) {
+ return getByteBufferHashCode((ByteBuffer)o);
+ }
+
+ return o.hashCode();
+ }*/
+
+ /** ByteBuffer.hashCode() takes into account the position of the
+ * buffer, but we do not want that*/
+ private int getByteBufferHashCode(ByteBuffer buf) {
+ int h = 1;
+ int p = buf.arrayOffset();
+ for (int j = buf.limit() - 1; j >= p; j--)
+ h = 31 * h + buf.get(j);
+ return h;
+ }
+
+ @Override
+ public Persistent clone() {
+ //return datumReader.clone(this, getSchema());
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append(super.toString());
+ builder.append(" {\n");
+ /*List<Field> fields = getSchema().getFields();
+ for(int i=0; i<fields.size(); i++) {
+ builder.append(" \"").append(fields.get(i).name()).append("\":\"");
+ builder.append(get(i)).append("\"\n");
+ }
+ */
+ builder.append("}");
+ return builder.toString();
+ }
+
+ protected boolean isFieldEqual(int index, Object value) {
+ /*Object old = get(index);
+ if (old == null && value == null)
+ return true;
+ if (old == null || value == null)
+ return false;
+ return value.equals(old);*/
+ return true;
+ }
+}
Added: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/StateManagerWSImpl.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/StateManagerWSImpl.java?rev=1356215&view=auto
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/StateManagerWSImpl.java (added)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/persistency/ws/impl/StateManagerWSImpl.java Mon Jul 2 13:38:43 2012
@@ -0,0 +1,104 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gora.persistency.ws.impl;
+
+import java.util.BitSet;
+
+import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.StateManager;
+
+/**
+ * An implementation for the StateManager. This implementation assumes
+ * every Persistent object has it's own StateManager.
+ */
+public class StateManagerWSImpl implements StateManager {
+
+ //TODO: serialize isNew in PersistentSerializer
+ protected boolean isNew;
+ protected BitSet dirtyBits;
+ protected BitSet readableBits;
+
+ public StateManagerWSImpl() {
+ }
+
+ public void setManagedPersistent(Persistent persistent) {
+ // dirtyBits = new BitSet(persistent.getSchema().getFields().size());
+ // readableBits = new BitSet(persistent.getSchema().getFields().size());
+ isNew = true;
+ }
+
+ @Override
+ public boolean isNew(Persistent persistent) {
+ return isNew;
+ }
+
+ @Override
+ public void setNew(Persistent persistent) {
+ this.isNew = true;
+ }
+
+ @Override
+ public void clearNew(Persistent persistent) {
+ this.isNew = false;
+ }
+
+ public void setDirty(Persistent persistent, int fieldIndex) {
+ dirtyBits.set(fieldIndex);
+ readableBits.set(fieldIndex);
+ }
+
+ public boolean isDirty(Persistent persistent, int fieldIndex) {
+ return dirtyBits.get(fieldIndex);
+ }
+
+ public boolean isDirty(Persistent persistent) {
+ return !dirtyBits.isEmpty();
+ }
+
+ @Override
+ public void setDirty(Persistent persistent) {
+ dirtyBits.set(0, dirtyBits.size());
+ }
+
+ @Override
+ public void clearDirty(Persistent persistent, int fieldIndex) {
+ dirtyBits.clear(fieldIndex);
+ }
+
+ public void clearDirty(Persistent persistent) {
+ dirtyBits.clear();
+ }
+
+ public void setReadable(Persistent persistent, int fieldIndex) {
+ readableBits.set(fieldIndex);
+ }
+
+ public boolean isReadable(Persistent persistent, int fieldIndex) {
+ return readableBits.get(fieldIndex);
+ }
+
+ @Override
+ public void clearReadable(Persistent persistent, int fieldIndex) {
+ readableBits.clear(fieldIndex);
+ }
+
+ public void clearReadable(Persistent persistent) {
+ readableBits.clear();
+ }
+}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/PartitionQuery.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/PartitionQuery.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/PartitionQuery.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/PartitionQuery.java Mon Jul 2 13:38:43 2012
@@ -18,14 +18,12 @@
package org.apache.gora.query;
-import org.apache.gora.persistency.Persistent;
-
/**
* PartitionQuery divides the results of the Query to multi partitions, so that
* queries can be run locally on the nodes that hold the data. PartitionQuery's are
* used for generating Hadoop InputSplits.
*/
-public interface PartitionQuery<K, T extends Persistent> extends Query<K, T> {
+public interface PartitionQuery<K, T> extends Query<K, T> {
/* PartitionQuery interface relaxes the dependency of DataStores to Hadoop*/
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Query.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Query.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Query.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Query.java Mon Jul 2 13:38:43 2012
@@ -20,16 +20,13 @@ package org.apache.gora.query;
import java.io.IOException;
-import org.apache.gora.persistency.Persistent;
import org.apache.gora.store.DataStore;
-import org.apache.hadoop.conf.Configurable;
-import org.apache.hadoop.io.Writable;
/**
* A query to a data store to retrieve objects. Queries are constructed by
* the DataStore implementation via {@link DataStore#newQuery()}.
*/
-public interface Query<K, T extends Persistent> extends Writable, Configurable {
+public interface Query<K, T> {
/**
* Sets the dataStore of this query. Under normal operation, this call
@@ -49,7 +46,7 @@ public interface Query<K, T extends Pers
* Executes the Query on the DataStore and returns the results.
* @return the {@link Result} for the query.
*/
- Result<K,T> execute() throws IOException;
+ Result<K,T> execute() throws Exception, IOException;
// /**
// * Compiles the query for performance and error checking. This
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Result.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Result.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Result.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/Result.java Mon Jul 2 13:38:43 2012
@@ -18,10 +18,8 @@
package org.apache.gora.query;
-import java.io.Closeable;
import java.io.IOException;
-import org.apache.gora.persistency.Persistent;
import org.apache.gora.store.DataStore;
/**
@@ -29,7 +27,7 @@ import org.apache.gora.store.DataStore;
* iterated by calling {@link #next()}, {@link #get()}
* and {@link #getKey()}.
*/
-public interface Result<K,T extends Persistent> extends Closeable {
+public interface Result<K,T> {
/**
* Returns the DataStore, that this Result is associated with.
@@ -47,7 +45,7 @@ public interface Result<K,T extends Pers
* Advances to the next element and returns false if at end.
* @return true if end is not reached yet
*/
- boolean next() throws IOException;
+ boolean next() throws Exception, IOException;
/**
* Returns the current key.
@@ -82,9 +80,8 @@ public interface Result<K,T extends Pers
/**
* Returns how far along the result has iterated, a value between 0 and 1.
*/
- float getProgress() throws IOException;
-
- @Override
+ float getProgress() throws IOException, InterruptedException, Exception;
+
void close() throws IOException;
}
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/FileSplitPartitionQuery.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/FileSplitPartitionQuery.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/FileSplitPartitionQuery.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/FileSplitPartitionQuery.java Mon Jul 2 13:38:43 2012
@@ -23,6 +23,7 @@ import java.io.DataOutput;
import java.io.IOException;
import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
@@ -31,7 +32,7 @@ import org.apache.hadoop.mapreduce.lib.i
* Keeps a {@link FileSplit} to represent the partition boundaries.
* FileSplitPartitionQuery is best used with existing {@link InputFormat}s.
*/
-public class FileSplitPartitionQuery<K, T extends Persistent>
+public class FileSplitPartitionQuery<K, T extends PersistentBase>
extends PartitionQueryImpl<K,T> {
private FileSplit split;
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/PartitionQueryImpl.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/PartitionQueryImpl.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/PartitionQueryImpl.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/PartitionQueryImpl.java Mon Jul 2 13:38:43 2012
@@ -23,16 +23,17 @@ import java.io.DataOutput;
import java.io.IOException;
import java.util.Arrays;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.PartitionQuery;
import org.apache.gora.query.Query;
import org.apache.gora.store.DataStore;
+import org.apache.gora.store.impl.DataStoreBase;
import org.apache.gora.util.IOUtils;
/**
* Implementation for {@link PartitionQuery}.
*/
-public class PartitionQueryImpl<K, T extends Persistent>
+public class PartitionQueryImpl<K, T extends PersistentBase>
extends QueryBase<K, T> implements PartitionQuery<K, T> {
protected Query<K, T> baseQuery;
@@ -53,7 +54,7 @@ public class PartitionQueryImpl<K, T ext
this.locations = locations;
setStartKey(startKey);
setEndKey(endKey);
- this.dataStore = baseQuery.getDataStore();
+ this.dataStore = (DataStoreBase<K, T>) baseQuery.getDataStore();
}
@Override
@@ -146,7 +147,7 @@ public String[] getLocations() {
//we should override the data store as basequery's data store
//also we may not call super.readFields so that temporary this.dataStore
//is not created at all
- this.dataStore = baseQuery.getDataStore();
+ this.dataStore = (DataStoreBase<K, T>) baseQuery.getDataStore();
}
@Override
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/QueryBase.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/QueryBase.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/QueryBase.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/QueryBase.java Mon Jul 2 13:38:43 2012
@@ -25,24 +25,27 @@ import java.io.IOException;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.lang.builder.ToStringBuilder;
-import org.apache.gora.persistency.Persistent;
+import org.apache.gora.persistency.impl.PersistentBase;
import org.apache.gora.query.Query;
import org.apache.gora.query.Result;
import org.apache.gora.store.DataStore;
+import org.apache.gora.store.impl.DataStoreBase;
import org.apache.gora.util.ClassLoadingUtils;
import org.apache.gora.util.IOUtils;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
/**
* Base class for Query implementations.
*/
-public abstract class QueryBase<K, T extends Persistent>
-implements Query<K,T> {
-
- protected DataStore<K,T> dataStore;
+public abstract class QueryBase<K, T extends PersistentBase>
+implements Query<K,T>, Writable, Configurable {
+
+ protected DataStoreBase<K,T> dataStore;
protected String queryString;
protected String[] fields;
@@ -62,11 +65,11 @@ implements Query<K,T> {
private Configuration conf;
public QueryBase(DataStore<K,T> dataStore) {
- this.dataStore = dataStore;
+ this.dataStore = (DataStoreBase<K, T>)dataStore;
}
@Override
- public Result<K,T> execute() throws IOException {
+ public Result<K,T> execute() throws Exception, IOException {
//compile();
return dataStore.execute(this);
}
@@ -80,7 +83,7 @@ implements Query<K,T> {
@Override
public void setDataStore(DataStore<K, T> dataStore) {
- this.dataStore = dataStore;
+ this.dataStore = (DataStoreBase<K, T>)dataStore;
}
@Override
@@ -203,12 +206,10 @@ public String[] getFields() {
return limit;
}
- @Override
public Configuration getConf() {
return conf;
}
- @Override
public void setConf(Configuration conf) {
this.conf = conf;
}
@@ -218,7 +219,7 @@ public String[] getFields() {
public void readFields(DataInput in) throws IOException {
String dataStoreClass = Text.readString(in);
try {
- dataStore = (DataStore<K, T>) ReflectionUtils.newInstance(ClassLoadingUtils.loadClass(dataStoreClass), conf);
+ dataStore = (DataStoreBase<K, T>) ReflectionUtils.newInstance(ClassLoadingUtils.loadClass(dataStoreClass), conf);
dataStore.readFields(in);
} catch (ClassNotFoundException ex) {
throw new IOException(ex);
@@ -242,7 +243,7 @@ public String[] getFields() {
limit = WritableUtils.readVLong(in);
}
- @Override
+ //@Override
public void write(DataOutput out) throws IOException {
//write datastore
Text.writeString(out, dataStore.getClass().getCanonicalName());
Modified: gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/ResultBase.java
URL: http://svn.apache.org/viewvc/gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/ResultBase.java?rev=1356215&r1=1356214&r2=1356215&view=diff
==============================================================================
--- gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/ResultBase.java (original)
+++ gora/branches/goraamazon/gora-core/src/main/java/org/apache/gora/query/impl/ResultBase.java Mon Jul 2 13:38:43 2012
@@ -101,17 +101,18 @@ public abstract class ResultBase<K, T ex
}
@Override
- public final boolean next() throws IOException {
- if(isLimitReached()) {
- return false;
- }
+ public final boolean next() throws Exception, IOException {
+ if(isLimitReached()) {
+ return false;
+ }
+
+ clear();
- clear();
- persistent = getOrCreatePersistent(persistent);
-
- boolean ret = nextInner();
- if(ret) ++offset;
- return ret;
+ persistent = getOrCreatePersistent(persistent);
+ boolean ret = nextInner();
+
+ if(ret) ++offset;
+ return ret;
}
@Override
@@ -125,10 +126,16 @@ public abstract class ResultBase<K, T ex
*/
protected abstract boolean nextInner() throws IOException;
- protected T getOrCreatePersistent(T persistent) throws IOException {
- if(persistent != null) {
- return persistent;
- }
- return dataStore.newPersistent();
+ protected T getOrCreatePersistent(T persistent) throws Exception, IOException {
+ if(persistent != null) {
+ return persistent;
+ }
+ return dataStore.newPersistent();
+ }
+
+ @Override
+ public void close() throws IOException{
+ // TODO Auto-generated method stub
+
}
}