You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2014/10/09 19:29:02 UTC
[1/3] Changes to repair to only bring us up to the current version
only. This is what we need in practice.
Repository: incubator-usergrid
Updated Branches:
refs/heads/collection_multiget c01b15044 -> f1a88aa90 (forced update)
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
index 3643bd1..ab0e2f4 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
@@ -17,6 +17,7 @@ import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.collection.mvcc.stage.AbstractIdStageTest;
import org.apache.usergrid.persistence.collection.mvcc.stage.CollectionIoEvent;
import org.apache.usergrid.persistence.collection.mvcc.stage.TestEntityGenerator;
+import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
import org.apache.usergrid.persistence.collection.service.UUIDService;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -46,6 +47,8 @@ public class LoadTest extends AbstractIdStageTest {
final CollectionScope collection = mock(CollectionScope.class);
final UUIDService uuidService = mock(UUIDService.class);
final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
+ final SerializationFig serializationFig = mock(SerializationFig.class);
+ when(serializationFig.getBufferSize()).thenReturn( 10 );
final UUID loadVersion = UUIDGenerator.newTimeUUID();
@@ -69,7 +72,7 @@ public class LoadTest extends AbstractIdStageTest {
- Load load = new Load( uuidService, serializationStrategy );
+ Load load = new Load( uuidService, serializationStrategy, serializationFig );
Entity loaded = load.call( entityIoEvent );
@@ -87,6 +90,9 @@ public class LoadTest extends AbstractIdStageTest {
final CollectionScope collection = mock(CollectionScope.class);
final UUIDService uuidService = mock(UUIDService.class);
final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
+ final SerializationFig serializationFig = mock(SerializationFig.class);
+ when(serializationFig.getBufferSize()).thenReturn( 10 );
+
final UUID loadVersion = UUIDGenerator.newTimeUUID();
@@ -128,7 +134,7 @@ public class LoadTest extends AbstractIdStageTest {
when(serializationStrategy.write( same(collection), writtenEntityArg.capture() )).thenReturn( batch );
- Load load = new Load( uuidService, serializationStrategy );
+ Load load = new Load( uuidService, serializationStrategy, serializationFig );
Entity loaded = load.call( entityIoEvent );
assertNotNull( loaded.getField( "derp" ) );
@@ -162,6 +168,9 @@ public class LoadTest extends AbstractIdStageTest {
final CollectionScope collection = mock(CollectionScope.class);
final UUIDService uuidService = mock(UUIDService.class);
final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
+ final SerializationFig serializationFig = mock(SerializationFig.class);
+ when(serializationFig.getBufferSize()).thenReturn( 10 );
+
final ChangeLogGenerator changeLogGenerator = new ChangeLogGeneratorImpl();
@@ -206,7 +215,7 @@ public class LoadTest extends AbstractIdStageTest {
when(serializationStrategy.write( same(collection), writtenEntityArg.capture() )).thenReturn( batch );
- Load load = new Load( uuidService, serializationStrategy );
+ Load load = new Load( uuidService, serializationStrategy, serializationFig );
Entity loaded = load.call( entityIoEvent );
assertNotNull( loaded.getField( "derple" ) );
@@ -236,6 +245,9 @@ public class LoadTest extends AbstractIdStageTest {
final CollectionScope collection = mock(CollectionScope.class);
final UUIDService uuidService = mock(UUIDService.class);
final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
+ final SerializationFig serializationFig = mock(SerializationFig.class);
+ when(serializationFig.getBufferSize()).thenReturn( 10 );
+
final UUID loadVersion = UUIDGenerator.newTimeUUID();
@@ -284,7 +296,7 @@ public class LoadTest extends AbstractIdStageTest {
when(serializationStrategy.write( same(collection), writtenEntityArg.capture() )).thenReturn( batch );
- Load load = new Load( uuidService, serializationStrategy );
+ Load load = new Load( uuidService, serializationStrategy, serializationFig );
Entity loaded = load.call( entityIoEvent );
assertNotNull( loaded.getField( "derp" ) );
@@ -317,6 +329,9 @@ public class LoadTest extends AbstractIdStageTest {
final CollectionScope collection = mock(CollectionScope.class);
final UUIDService uuidService = mock(UUIDService.class);
final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
+ final SerializationFig serializationFig = mock(SerializationFig.class);
+ when(serializationFig.getBufferSize()).thenReturn( 10 );
+
final UUID loadVersion = UUIDGenerator.newTimeUUID();
@@ -334,7 +349,7 @@ public class LoadTest extends AbstractIdStageTest {
//mock up returning a list of MvccEntities
when(serializationStrategy.load( collection, entityId, loadVersion, 1 )).thenReturn( results);
- Load load = new Load( uuidService, serializationStrategy );
+ Load load = new Load( uuidService, serializationStrategy, serializationFig );
Entity loaded = load.call( entityIoEvent );
assertNull( "No entity was loaded", loaded );
@@ -347,6 +362,9 @@ public class LoadTest extends AbstractIdStageTest {
@Override
protected void validateStage( final CollectionIoEvent<Id> event ) {
final UUIDService uuidService = mock(UUIDService.class);
+ final SerializationFig serializationFig = mock(SerializationFig.class);
+ when(serializationFig.getBufferSize()).thenReturn( 10 );
+
/**
* Mock up wrong UUID type
@@ -356,6 +374,6 @@ public class LoadTest extends AbstractIdStageTest {
final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
- new Load(uuidService, serializationStrategy).call( event );
+ new Load(uuidService, serializationStrategy, serializationFig).call( event );
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
new file mode 100644
index 0000000..f4a75d4
--- /dev/null
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
@@ -0,0 +1,136 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. 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. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+
+package org.apache.usergrid.persistence.collection.serialization;
+
+
+import java.util.Arrays;
+import java.util.UUID;
+
+import org.junit.Test;
+
+import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.impl.CollectionScopeImpl;
+import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
+import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
+import org.apache.usergrid.persistence.collection.mvcc.entity.impl.MvccEntityImpl;
+import org.apache.usergrid.persistence.collection.serialization.impl.EntityRepairImpl;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
+import org.apache.usergrid.persistence.model.entity.SimpleId;
+import org.apache.usergrid.persistence.model.field.StringField;
+import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+/**
+ * Tests the entity repair task
+ */
+public class EntityRepairImplTest {
+
+
+ /**
+ * Tests changing from a full version to 2 updates, ensures we have a proper output
+ */
+ @Test
+ public void testSimpleRolling() {
+
+ final SerializationFig serializationFig = mock( SerializationFig.class );
+
+ when( serializationFig.getBufferSize() ).thenReturn( 10 );
+
+
+ final Id simpleId = new SimpleId( "entity" );
+
+ final Entity v1Entity = new Entity( simpleId );
+
+
+ v1Entity.setField( new StringField( "field1", "value1" ) );
+ v1Entity.setField( new StringField( "field2", "value2" ) );
+
+
+ final MvccEntityImpl v1 =
+ new MvccEntityImpl( simpleId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.COMPLETE, v1Entity );
+
+
+ final Entity v2Entity = new Entity( simpleId );
+ v2Entity.setField( new StringField( "field1", "value1.1" ) );
+
+ final MvccEntityImpl v2 =
+ new MvccEntityImpl( simpleId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, v2Entity );
+
+
+ final Entity v3Entity = new Entity( simpleId );
+ v3Entity.setField( new StringField( "field2", "value1.2" ) );
+
+ final MvccEntityImpl v3 =
+ new MvccEntityImpl( simpleId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, v3Entity );
+
+
+ final MvccEntitySerializationStrategy mvccEntitySerializationStrategy =
+ mock( MvccEntitySerializationStrategy.class );
+
+
+ final Id applicationId = new SimpleId( "application" );
+
+ final CollectionScope scope = new CollectionScopeImpl( applicationId, applicationId, "users" );
+
+ //mock up returning
+ when( mvccEntitySerializationStrategy
+ .load( scope, simpleId, v3.getVersion(), serializationFig.getBufferSize() ) )
+ .thenReturn( Arrays.<MvccEntity>asList( v3, v2, v1 ).iterator() );
+
+
+ EntityRepairImpl entityRepair = new EntityRepairImpl( mvccEntitySerializationStrategy, serializationFig );
+
+ final MvccEntity returned = entityRepair.maybeRepair( scope, v3 );
+
+ final UUID version = returned.getVersion();
+
+ assertEquals( "Versions should match", v3.getVersion(), version );
+
+
+
+ final Id entityId = returned.getId();
+
+ assertEquals( "Entity Id's match", simpleId, entityId );
+
+
+
+ final Entity finalVersion = returned.getEntity().get();
+
+ final Object expectedField1Value = v2.getEntity().get().getField( "field1" ).getValue();
+
+ final Object returnedField1Value = finalVersion.getField( "field1" ).getValue();
+
+ assertEquals( "Same field value", expectedField1Value, returnedField1Value );
+
+
+
+ final Object expectedField2Value = v3.getEntity().get().getField( "field2" ).getValue();
+
+ final Object returnedField2Value = finalVersion.getField( "field2" ).getValue();
+
+ assertEquals( "Same field value", expectedField2Value, returnedField2Value );
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
index 6780115..ba5ebce 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
@@ -649,7 +649,7 @@ public class MvccEntitySerializationStrategyImplTest {
public void loadParamEntityId() throws ConnectionException {
serializationStrategy
- .load( new CollectionScopeImpl(new SimpleId( "organization" ), new SimpleId( "test" ), "test" ), null, UUIDGenerator.newTimeUUID() );
+ .load( new CollectionScopeImpl(new SimpleId( "organization" ), new SimpleId( "test" ), "test" ), (Id)null, UUIDGenerator.newTimeUUID() );
}
[3/3] git commit: Refactored read stage
Posted by to...@apache.org.
Refactored read stage
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/f1a88aa9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/f1a88aa9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/f1a88aa9
Branch: refs/heads/collection_multiget
Commit: f1a88aa908211e0bb2c137f96c2fe0028ff202ac
Parents: f7d406e
Author: Todd Nine <to...@apache.org>
Authored: Thu Oct 9 11:27:23 2014 -0600
Committer: Todd Nine <to...@apache.org>
Committed: Thu Oct 9 11:28:56 2014 -0600
----------------------------------------------------------------------
.../persistence/collection/EntitySet.java | 3 +-
.../impl/EntityCollectionManagerImpl.java | 45 ++-
.../mvcc/MvccEntitySerializationStrategy.java | 12 -
.../collection/mvcc/stage/load/Load.java | 111 ------
.../serialization/SerializationFig.java | 19 +
.../serialization/impl/EntityRepairImpl.java | 5 +-
.../serialization/impl/EntitySetImpl.java | 10 +-
.../MvccEntitySerializationStrategyImpl.java | 32 +-
.../collection/EntityCollectionManagerIT.java | 240 +++++++++---
.../collection/mvcc/stage/load/LoadTest.java | 379 -------------------
.../serialization/EntityRepairImplTest.java | 25 +-
...MvccEntitySerializationStrategyImplTest.java | 40 +-
12 files changed, 290 insertions(+), 631 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
index 32c168f..7803a04 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
@@ -18,6 +18,7 @@ package org.apache.usergrid.persistence.collection;/*
*/
+import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -32,7 +33,7 @@ public interface EntitySet {
* @param entityId
* @return
*/
- public Entity getEntity(Id entityId);
+ public MvccEntity getEntity(Id entityId);
/**
* Get the number of entities in this set
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
index 4a72285..39df9cb 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
@@ -20,6 +20,7 @@ package org.apache.usergrid.persistence.collection.impl;
import java.util.Collection;
+import java.util.Collections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,12 +31,12 @@ import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
import org.apache.usergrid.persistence.collection.guice.Write;
import org.apache.usergrid.persistence.collection.guice.WriteUpdate;
+import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccValidationUtils;
import org.apache.usergrid.persistence.collection.mvcc.stage.CollectionIoEvent;
import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkCommit;
import org.apache.usergrid.persistence.collection.mvcc.stage.delete.MarkStart;
-import org.apache.usergrid.persistence.collection.mvcc.stage.load.Load;
import org.apache.usergrid.persistence.collection.mvcc.stage.write.RollbackAction;
import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteCommit;
import org.apache.usergrid.persistence.collection.mvcc.stage.write.WriteOptimisticVerify;
@@ -46,6 +47,7 @@ import org.apache.usergrid.persistence.core.task.TaskExecutor;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
+import org.apache.usergrid.persistence.model.util.UUIDGenerator;
import com.google.common.base.Preconditions;
import com.google.inject.Inject;
@@ -79,9 +81,6 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
private final WriteCommit writeCommit;
private final RollbackAction rollback;
- //load stages
- private final Load load;
-
//delete stages
private final MarkStart markStart;
@@ -89,16 +88,20 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
private final TaskExecutor taskExecutor;
+ private final MvccEntitySerializationStrategy entitySerializationStrategy;
+
@Inject
public EntityCollectionManagerImpl( final UUIDService uuidService, @Write final WriteStart writeStart,
@WriteUpdate final WriteStart writeUpdate,
final WriteUniqueVerify writeVerifyUnique,
final WriteOptimisticVerify writeOptimisticVerify,
- final WriteCommit writeCommit, final RollbackAction rollback, final Load load,
+ final WriteCommit writeCommit, final RollbackAction rollback,
final MarkStart markStart, final MarkCommit markCommit,
- @CollectionTaskExecutor
- final TaskExecutor taskExecutor,
- @Assisted final CollectionScope collectionScope) {
+ final MvccEntitySerializationStrategy entitySerializationStrategy,
+ @CollectionTaskExecutor final TaskExecutor taskExecutor,
+ @Assisted final CollectionScope collectionScope
+ ) {
+
Preconditions.checkNotNull( uuidService, "uuidService must be defined" );
@@ -111,13 +114,13 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
this.writeCommit = writeCommit;
this.rollback = rollback;
- this.load = load;
this.markStart = markStart;
this.markCommit = markCommit;
this.uuidService = uuidService;
this.collectionScope = collectionScope;
this.taskExecutor = taskExecutor;
+ this.entitySerializationStrategy = entitySerializationStrategy;
}
@@ -183,14 +186,30 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
Preconditions.checkNotNull( entityId.getUuid(), "Entity id uuid required in load stage" );
Preconditions.checkNotNull( entityId.getType(), "Entity id type required in load stage" );
- return Observable.from( new CollectionIoEvent<Id>( collectionScope, entityId ) )
- .map( load );
+ return load( Collections.singleton(entityId) ).map( new Func1<EntitySet, Entity>() {
+ @Override
+ public Entity call( final EntitySet entitySet ) {
+ final MvccEntity entity = entitySet.getEntity( entityId );
+
+ if(entity == null){
+ return null;
+ }
+
+ return entity.getEntity().orNull();
+ }
+ } );
}
@Override
public Observable<EntitySet> load( final Collection<Id> entityIds ) {
- return null;
+
+ Preconditions.checkNotNull( entityIds, "entityIds cannot be null" );
+
+ final EntitySet
+ results = entitySerializationStrategy.load( collectionScope, entityIds, UUIDGenerator.newTimeUUID() );
+
+ return Observable.just( results );
}
@@ -222,7 +241,7 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
//we an update, signal the fix
//TODO T.N Change this to fire a task
- Observable.from( new CollectionIoEvent<Id>(collectionScope, entityId ) ).map( load ).subscribeOn( Schedulers.io() ).subscribe();
+// Observable.from( new CollectionIoEvent<Id>(collectionScope, entityId ) ).map( load ).subscribeOn( Schedulers.io() ).subscribe();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
index 7947029..0467c95 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
@@ -47,18 +47,6 @@ public interface MvccEntitySerializationStrategy {
public MutationBatch write( CollectionScope context, MvccEntity entity );
- /**
- * Load and return the entity with the given id and a version that is <= the version provided
- *
- * @param context The context to persist the entity into
- * @param entityId The entity id to load
- * @param version The version to load. This will return the version <= the given version
- *
- * @return The deserialized version of the entity. Null if no version == to version exists. If the entity version
- * has been cleared, the MvccEntity will be returned, but the optional entity will not be set
- */
- public MvccEntity load( CollectionScope context, Id entityId, UUID version );
-
/**
* Load the entities into the entitySet from the specified Ids. Loads versions <= the maxVersion
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
deleted file mode 100644
index f9c66cc..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.usergrid.persistence.collection.mvcc.stage.load;
-
-
-import java.util.Iterator;
-import java.util.UUID;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
-import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
-import org.apache.usergrid.persistence.collection.mvcc.stage.CollectionIoEvent;
-import org.apache.usergrid.persistence.collection.serialization.EntityRepair;
-import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
-import org.apache.usergrid.persistence.collection.serialization.impl.EntityRepairImpl;
-import org.apache.usergrid.persistence.collection.service.UUIDService;
-import org.apache.usergrid.persistence.core.util.ValidationUtils;
-import org.apache.usergrid.persistence.model.entity.Entity;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.google.common.base.Preconditions;
-import com.google.inject.Inject;
-import com.google.inject.Singleton;
-
-import rx.functions.Func1;
-
-
-/**
- * This stage is a load stage to load a single entity
- */
-@Singleton
-public class Load implements Func1<CollectionIoEvent<Id>, Entity> {
-
-
- private static final Logger LOG = LoggerFactory.getLogger( Load.class );
-
- private final UUIDService uuidService;
- private final MvccEntitySerializationStrategy entitySerializationStrategy;
- private final EntityRepair entityRepair;
-
-
- @Inject
- public Load( final UUIDService uuidService, final MvccEntitySerializationStrategy entitySerializationStrategy, final
- SerializationFig serializationFig ) {
- Preconditions.checkNotNull( entitySerializationStrategy, "entitySerializationStrategy is required" );
- Preconditions.checkNotNull( uuidService, "uuidService is required" );
-
-
- this.uuidService = uuidService;
- this.entitySerializationStrategy = entitySerializationStrategy;
- entityRepair = new EntityRepairImpl( entitySerializationStrategy, serializationFig );
-
- }
-
- //TODO: do reads partial merges in batches. maybe 5 or 10 at a time.
- /**
- * for example
- so if like v1 is a full
- and you have v1 -> v20, where v2->20 is all partial
- you merge up to 10, then flush
- then process 10->20, then flush
- */
- @Override
- public Entity call( final CollectionIoEvent<Id> idIoEvent ) {
- final Id entityId = idIoEvent.getEvent();
-
- ValidationUtils.verifyIdentity( entityId );
-
-
- final CollectionScope collectionScope = idIoEvent.getEntityCollection();
-
- //generate a version that represents now
- final UUID versionMax = uuidService.newTimeUUID();
-
- Iterator<MvccEntity> results = entitySerializationStrategy.load(
- collectionScope, entityId, versionMax, 1 );
-
- if(!results.hasNext()){
- return null;
- }
-
- final MvccEntity returned = results.next();
-
- final MvccEntity repairedEntity = entityRepair.maybeRepair( collectionScope, returned );
-
- if(repairedEntity == null)
- return null;
-
- return repairedEntity.getEntity().orNull();
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/SerializationFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/SerializationFig.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/SerializationFig.java
index 7e69a19..15bae5c 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/SerializationFig.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/SerializationFig.java
@@ -56,4 +56,23 @@ public interface SerializationFig extends GuicyFig {
@Key( "collection.task.pool.queuesize" )
@Default( "20" )
int getTaskPoolQueueSize();
+
+ /**
+ * The maximum amount of entites we can load at any one time
+ * @return
+ */
+ @Key( "collection.max.load.size" )
+ @Default( "1000" )
+ int getMaxLoadSize();
+
+
+
+ /**
+ * The maximum number of bytes a serialized entity can be. Any thing beyond this is rejected
+ * @return
+ */
+ @Key( "collection.max.entry.size" )
+ @Default( "512" )
+ int getMaxEntrySize();
+
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
index d26cd3f..62f4915 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
@@ -36,6 +36,7 @@ import org.apache.usergrid.persistence.model.field.Field;
import com.google.inject.Inject;
import com.google.inject.Singleton;
+import com.netflix.astyanax.MutationBatch;
@Singleton
@@ -102,7 +103,9 @@ public class EntityRepairImpl implements EntityRepair {
final MvccEntity mergedEntity = entityRepair( changeLog, targetEntity );
try {
- mvccEntitySerializationStrategy.write( collectionScope, mergedEntity ).execute();
+ final MutationBatch batch = mvccEntitySerializationStrategy.write( collectionScope, mergedEntity );
+
+ batch.execute();
}
catch ( Exception e ) {
throw new RuntimeException( "Couldn't rewrite repaired entity", e );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
index 89ec69e..9e03754 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
@@ -45,14 +45,8 @@ public class EntitySetImpl implements EntitySet {
@Override
- public Entity getEntity( final Id entityId ) {
- final MvccEntity entity = entities.get( entityId );
-
- if ( entity == null ) {
- return null;
- }
-
- return entity.getEntity().orNull();
+ public MvccEntity getEntity( final Id entityId ) {
+ return entities.get( entityId );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
index b36ee0e..5449e46 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
@@ -142,33 +142,6 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
}
- @Override
- public MvccEntity load( final CollectionScope collectionScope, final Id entityId, final UUID version ) {
- Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
- Preconditions.checkNotNull( entityId, "entity id is required" );
- Preconditions.checkNotNull( version, "version is required" );
-
-
- Column<UUID> column;
-
- try {
- column = keyspace.prepareQuery( CF_ENTITY_DATA ).getKey( ScopedRowKey.fromKey( collectionScope, entityId ) )
- .getColumn( version ).execute().getResult();
- }
-
- catch ( NotFoundException e ) {
- //swallow, there's just no column
- return null;
- }
- catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope, "An error occurred connecting to cassandra",
- e );
- }
-
-
- return new MvccColumnParser( entityId ).parseColumn( column );
- }
-
@Override
public EntitySet load( final CollectionScope collectionScope, final Collection<Id> entityIds,
@@ -181,6 +154,11 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
Preconditions.checkNotNull( maxVersion, "version is required" );
+ //didnt put the max in the error message, I don't want to take the string construction hit every time
+ Preconditions.checkArgument( entityIds.size() <= serializationFig.getMaxLoadSize(), "requested size cannot be over configured maximum");
+
+
+
final List<ScopedRowKey<CollectionScope, Id>> rowKeys = new ArrayList<>( entityIds.size() );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
index 9e8bbe3..823cb87 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/EntityCollectionManagerIT.java
@@ -18,6 +18,8 @@
package org.apache.usergrid.persistence.collection;
+import java.util.ArrayList;
+import java.util.List;
import java.util.UUID;
import org.jukito.UseModules;
@@ -29,20 +31,27 @@ import org.apache.usergrid.persistence.collection.exception.WriteUniqueVerifyExc
import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
import org.apache.usergrid.persistence.collection.guice.TestCollectionModule;
import org.apache.usergrid.persistence.collection.impl.CollectionScopeImpl;
+import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
+import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
import org.apache.usergrid.persistence.core.cassandra.ITRunner;
import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
+import org.apache.usergrid.persistence.model.field.BooleanField;
import org.apache.usergrid.persistence.model.field.IntegerField;
import org.apache.usergrid.persistence.model.field.StringField;
import com.fasterxml.uuid.UUIDComparator;
-import com.fasterxml.uuid.impl.UUIDUtil;
import com.google.inject.Inject;
import rx.Observable;
-
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/** @author tnine */
@@ -57,12 +66,15 @@ public class EntityCollectionManagerIT {
@Rule
public MigrationManagerRule migrationManagerRule;
+ @Inject
+ private SerializationFig serializationFig;
+
@Test
public void write() {
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
Entity newEntity = new Entity( new SimpleId( "test" ) );
@@ -78,17 +90,18 @@ public class EntityCollectionManagerIT {
assertNotNull( "Version exists", returned.getVersion() );
}
+
@Test
public void writeWithUniqueValues() {
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
EntityCollectionManager manager = factory.createCollectionManager( context );
{
Entity newEntity = new Entity( new SimpleId( "test" ) );
- newEntity.setField( new IntegerField("count", 5, true) );
+ newEntity.setField( new IntegerField( "count", 5, true ) );
Observable<Entity> observable = manager.write( newEntity );
Entity returned = observable.toBlocking().lastOrDefault( null );
@@ -97,23 +110,24 @@ public class EntityCollectionManagerIT {
{
try {
Entity newEntity = new Entity( new SimpleId( "test" ) );
- newEntity.setField( new IntegerField("count", 5, true) );
+ newEntity.setField( new IntegerField( "count", 5, true ) );
manager.write( newEntity ).toBlocking().last();
- fail("Write should have thrown an exception");
-
- } catch ( Exception ex ) {
- WriteUniqueVerifyException e = (WriteUniqueVerifyException)ex;
+ fail( "Write should have thrown an exception" );
+ }
+ catch ( Exception ex ) {
+ WriteUniqueVerifyException e = ( WriteUniqueVerifyException ) ex;
assertEquals( 1, e.getVioliations().size() );
}
}
}
+
@Test
public void writeAndLoad() {
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
Entity newEntity = new Entity( new SimpleId( "test" ) );
@@ -139,8 +153,8 @@ public class EntityCollectionManagerIT {
@Test
public void writeLoadDelete() {
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
Entity newEntity = new Entity( new SimpleId( "test" ) );
EntityCollectionManager manager = factory.createCollectionManager( context );
@@ -174,8 +188,8 @@ public class EntityCollectionManagerIT {
@Test
public void writeLoadUpdateLoad() {
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
Entity newEntity = new Entity( new SimpleId( "test" ) );
newEntity.setField( new IntegerField( "counter", 1 ) );
@@ -221,8 +235,8 @@ public class EntityCollectionManagerIT {
public void writeAndLoadScopeClosure() {
- CollectionScope collectionScope1 = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test1" ), "test1" );
+ CollectionScope collectionScope1 =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test1" ), "test1" );
Entity newEntity = new Entity( new SimpleId( "test" ) );
@@ -245,30 +259,32 @@ public class EntityCollectionManagerIT {
//now make sure we can't load it from another scope, using the same org
- CollectionScope collectionScope2 = new CollectionScopeImpl(
- collectionScope1.getApplication(), new SimpleId("test2"), collectionScope1.getName());
+ CollectionScope collectionScope2 =
+ new CollectionScopeImpl( collectionScope1.getApplication(), new SimpleId( "test2" ),
+ collectionScope1.getName() );
EntityCollectionManager manager2 = factory.createCollectionManager( collectionScope2 );
- Entity loaded = manager2.load( createReturned.getId() )
- .toBlocking().lastOrDefault( null );
+ Entity loaded = manager2.load( createReturned.getId() ).toBlocking().lastOrDefault( null );
assertNull( "CollectionScope works correctly", loaded );
//now try to load it from another org, with the same scope
- CollectionScope collectionScope3 = new CollectionScopeImpl(
- new SimpleId("organization2"), collectionScope1.getOwner(), collectionScope1.getName());
+ CollectionScope collectionScope3 =
+ new CollectionScopeImpl( new SimpleId( "organization2" ), collectionScope1.getOwner(),
+ collectionScope1.getName() );
assertNotNull( collectionScope3 );
}
+
@Test
public void partialUpdate() {
- StringField testField1 = new StringField("testField","value");
+ StringField testField1 = new StringField( "testField", "value" );
StringField addedField = new StringField( "testFud", "NEWPARTIALUPDATEZOMG" );
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "testUpdate" ), "testUpdate" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "testUpdate" ), "testUpdate" );
Entity oldEntity = new Entity( new SimpleId( "testUpdate" ) );
oldEntity.setField( new StringField( "testField", "value" ) );
@@ -283,12 +299,12 @@ public class EntityCollectionManagerIT {
final UUID writeVersion = returned.getVersion();
- assertNotNull("Write version was set", writeVersion);
+ assertNotNull( "Write version was set", writeVersion );
/**
* Modify the oldEntity
*/
- oldEntity.getFields().remove( testField1 );
+ oldEntity.getFields().remove( testField1 );
oldEntity.setField( addedField );
observable = manager.update( oldEntity );
@@ -300,7 +316,7 @@ public class EntityCollectionManagerIT {
final UUID updatedVersion = updateReturned.getVersion();
- assertNotNull("Updated version returned", updatedVersion);
+ assertNotNull( "Updated version returned", updatedVersion );
assertTrue( "Updated version higher", UUIDComparator.staticCompare( updatedVersion, writeVersion ) > 0 );
@@ -309,21 +325,20 @@ public class EntityCollectionManagerIT {
final UUID returnedVersion = newEntity.getVersion();
- assertEquals("Loaded version matches updated version", updatedVersion, returnedVersion);
+ assertEquals( "Loaded version matches updated version", updatedVersion, returnedVersion );
assertNotNull( "Returned has a uuid", returned.getId() );
- assertEquals( addedField, newEntity.getField( "testFud" ));
-
-
+ assertEquals( addedField, newEntity.getField( "testFud" ) );
}
+
@Test
public void partialUpdateDelete() {
- StringField testField = new StringField("testField","value");
+ StringField testField = new StringField( "testField", "value" );
StringField addedField = new StringField( "testFud", "NEWPARTIALUPDATEZOMG" );
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "testUpdate" ), "testUpdate" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "testUpdate" ), "testUpdate" );
Entity oldEntity = new Entity( new SimpleId( "testUpdate" ) );
oldEntity.setField( new StringField( "testField", "value" ) );
@@ -336,13 +351,13 @@ public class EntityCollectionManagerIT {
assertNotNull( "Returned has a uuid", returned.getId() );
- oldEntity.getFields().remove( testField );
+ oldEntity.getFields().remove( testField );
oldEntity.setField( addedField );
//Entity is deleted then updated right afterwards.
manager.delete( oldEntity.getId() );
- observable = manager.update( oldEntity);
+ observable = manager.update( oldEntity );
returned = observable.toBlocking().lastOrDefault( null );
@@ -353,9 +368,7 @@ public class EntityCollectionManagerIT {
Entity newEntity = newEntityObs.toBlocking().last();
assertNotNull( "Returned has a uuid", returned.getId() );
- assertEquals( addedField, newEntity.getField( addedField.getName()) );
-
-
+ assertEquals( addedField, newEntity.getField( addedField.getName() ) );
}
@@ -366,10 +379,10 @@ public class EntityCollectionManagerIT {
Entity origEntity = new Entity( new SimpleId( "testUpdate" ) );
origEntity.setField( new StringField( "testField", "value" ) );
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "testUpdate" ), "testUpdate" );
+ CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "testUpdate" ), "testUpdate" );
EntityCollectionManager manager = factory.createCollectionManager( context );
- Entity returned = manager.write(origEntity).toBlocking().lastOrDefault( null );
+ Entity returned = manager.write( origEntity ).toBlocking().lastOrDefault( null );
// note its version
UUID oldVersion = returned.getVersion();
@@ -379,17 +392,140 @@ public class EntityCollectionManagerIT {
// partial update entity but we don't have version number
Entity updateEntity = new Entity( origEntity.getId() );
- updateEntity.setField( new StringField("addedField", "other value" ) );
- manager.update(origEntity).toBlocking().lastOrDefault(null);
+ updateEntity.setField( new StringField( "addedField", "other value" ) );
+ manager.update( origEntity ).toBlocking().lastOrDefault( null );
// get entity now, it must have a new version
- returned = manager.load(origEntity.getId() ).toBlocking().lastOrDefault( null );
+ returned = manager.load( origEntity.getId() ).toBlocking().lastOrDefault( null );
UUID newVersion = returned.getVersion();
assertNotNull( "A new version must be assigned", newVersion );
// new Version should be > old version
- assertTrue( UUIDComparator.staticCompare( newVersion, oldVersion ) > 0);
+ assertTrue( UUIDComparator.staticCompare( newVersion, oldVersion ) > 0 );
+ }
+
+
+ @Test
+ public void writeMultiget() {
+
+ final CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ final EntityCollectionManager manager = factory.createCollectionManager( context );
+
+ final int multigetSize = serializationFig.getMaxLoadSize();
+
+ final List<Entity> writtenEntities = new ArrayList<>( multigetSize );
+ final List<Id> entityIds = new ArrayList<>( multigetSize );
+
+ for ( int i = 0; i < multigetSize; i++ ) {
+ final Entity entity = new Entity( new SimpleId( "test" ) );
+
+ final Entity written = manager.write( entity ).toBlocking().last();
+
+ writtenEntities.add( written );
+ entityIds.add( written.getId() );
+ }
+
+
+ final EntitySet entitySet = manager.load( entityIds ).toBlocking().lastOrDefault( null );
+
+ assertNotNull( entitySet );
+
+ assertEquals(multigetSize, entitySet.size());
+ assertFalse(entitySet.isEmpty());
+
+ /**
+ * Validate every element exists
+ */
+ for(int i = 0; i < multigetSize; i ++){
+ final Entity expected = writtenEntities.get( i );
+
+ final MvccEntity returned = entitySet.getEntity( expected.getId() );
+
+ assertEquals("Same entity returned", expected, returned.getEntity().get());
+ }
+
+
}
+
+ /**
+ * Perform a multiget where every entity will need repaired on load
+ */
+ @Test
+ public void writeMultigetRepair() {
+
+ final CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ final EntityCollectionManager manager = factory.createCollectionManager( context );
+
+ final int multigetSize = serializationFig.getMaxLoadSize();
+
+ final List<Entity> writtenEntities = new ArrayList<>( multigetSize );
+ final List<Id> entityIds = new ArrayList<>( multigetSize );
+
+ for ( int i = 0; i < multigetSize; i++ ) {
+ final Entity entity = new Entity( new SimpleId( "test" ) );
+
+ final Entity written = manager.write( entity ).toBlocking().last();
+
+ written.setField( new BooleanField( "updated", true ) );
+
+ final Entity updated = manager.update( written ).toBlocking().last();
+
+ writtenEntities.add( updated );
+ entityIds.add( updated.getId() );
+ }
+
+
+ final EntitySet entitySet = manager.load( entityIds ).toBlocking().lastOrDefault( null );
+
+ assertNotNull( entitySet );
+
+ assertEquals(multigetSize, entitySet.size());
+ assertFalse(entitySet.isEmpty());
+
+ /**
+ * Validate every element exists
+ */
+ for(int i = 0; i < multigetSize; i ++){
+ final Entity expected = writtenEntities.get( i );
+
+ final MvccEntity returned = entitySet.getEntity( expected.getId() );
+
+ assertEquals("Same entity returned", expected, returned.getEntity().get());
+
+ assertTrue( ( Boolean ) returned.getEntity().get().getField( "updated" ).getValue() );
+ }
+
+
+ }
+
+
+
+ @Test(expected = IllegalArgumentException.class)
+ public void readTooLarge() {
+
+ final CollectionScope context =
+ new CollectionScopeImpl( new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ final EntityCollectionManager manager = factory.createCollectionManager( context );
+
+ final int multigetSize = serializationFig.getMaxLoadSize() +1;
+
+
+ final List<Id> entityIds = new ArrayList<>( multigetSize );
+
+ for ( int i = 0; i < multigetSize; i++ ) {
+
+ entityIds.add( new SimpleId( "simple" ) );
+ }
+
+
+ //should throw an exception
+ manager.load( entityIds ).toBlocking().lastOrDefault( null );
+
+
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
deleted file mode 100644
index ab0e2f4..0000000
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/LoadTest.java
+++ /dev/null
@@ -1,379 +0,0 @@
-package org.apache.usergrid.persistence.collection.mvcc.stage.load;
-
-
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.UUID;
-
-import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-
-import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
-import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGenerator;
-import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGeneratorImpl;
-import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
-import org.apache.usergrid.persistence.collection.mvcc.stage.AbstractIdStageTest;
-import org.apache.usergrid.persistence.collection.mvcc.stage.CollectionIoEvent;
-import org.apache.usergrid.persistence.collection.mvcc.stage.TestEntityGenerator;
-import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
-import org.apache.usergrid.persistence.collection.service.UUIDService;
-import org.apache.usergrid.persistence.model.entity.Entity;
-import org.apache.usergrid.persistence.model.entity.Id;
-import org.apache.usergrid.persistence.model.field.StringField;
-import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-
-import com.google.common.collect.Lists;
-import com.netflix.astyanax.MutationBatch;
-import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.mockito.Matchers.same;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-
-/** @author tnine */
-
-public class LoadTest extends AbstractIdStageTest {
-
-
- @Test
- public void testLoadWithData(){
- final CollectionScope collection = mock(CollectionScope.class);
- final UUIDService uuidService = mock(UUIDService.class);
- final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
- final SerializationFig serializationFig = mock(SerializationFig.class);
- when(serializationFig.getBufferSize()).thenReturn( 10 );
-
-
- final UUID loadVersion = UUIDGenerator.newTimeUUID();
-
- //mock up the time uuid
- when(uuidService.newTimeUUID()).thenReturn(loadVersion);
-
- final Id entityId = TestEntityGenerator.generateId();
-
- final CollectionIoEvent<Id> entityIoEvent = new CollectionIoEvent<Id>(collection, entityId );
-
-
- final Entity entity = TestEntityGenerator.generateEntity(entityId, loadVersion);
-
- final MvccEntity mvccEntity = TestEntityGenerator.fromEntityStatus( entity, MvccEntity.Status.COMPLETE );
-
- final Iterator<MvccEntity> results = Lists.newArrayList( mvccEntity ).iterator();
-
- //mock up returning a list of MvccEntities
- when(serializationStrategy.load( collection, entityId, loadVersion, 1 )).thenReturn( results);
-
-
-
- Load load = new Load( uuidService, serializationStrategy, serializationFig );
- Entity loaded = load.call( entityIoEvent );
-
-
- assertSame("Same entity was loaded", entity, loaded);
-
- }
-
-
- /**
- * Handles second trigger condition with partial updates.
- * A read on an entity , and we recognize that the entity we are reading is partial.
- */
- @Test
- public void testLoadWithPartialWrite() throws ConnectionException {
- final CollectionScope collection = mock(CollectionScope.class);
- final UUIDService uuidService = mock(UUIDService.class);
- final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
- final SerializationFig serializationFig = mock(SerializationFig.class);
- when(serializationFig.getBufferSize()).thenReturn( 10 );
-
-
-
- final UUID loadVersion = UUIDGenerator.newTimeUUID();
-
- //mock up the time uuid
- when(uuidService.newTimeUUID()).thenReturn(loadVersion);
-
- final Id entityId = TestEntityGenerator.generateId();
-
- final CollectionIoEvent<Id> entityIoEvent = new CollectionIoEvent<Id>(collection, entityId );
-
-
- final Entity entity = TestEntityGenerator.generateEntity(entityId, loadVersion);
- entity.setField( new StringField( "derp","noderp" ) );
-
- final MvccEntity completeMvccEntity = TestEntityGenerator.fromEntityStatus( entity, MvccEntity.Status.COMPLETE );
-
-
- final Entity entity2 = TestEntityGenerator.generateEntity( entityId, UUIDGenerator.newTimeUUID() );
- entity2.setField( new StringField( "derp","noderp" ) );
- entity2.setField( new StringField( "merple","nomerple" ) );
-
- final MvccEntity partialMvccEntity = TestEntityGenerator.fromEntityStatus( entity2, MvccEntity.Status.PARTIAL );
-
- final List<MvccEntity> results = Lists.newArrayList( partialMvccEntity );
- results.add( completeMvccEntity );
-
-
- //mock up returning a list of MvccEntities
- when( serializationStrategy.load( collection, entityId, loadVersion, 1 ) ).thenReturn( results.iterator());
-
- /**
- * Mock up returning the batch for repair
- */
-
- final MutationBatch batch = mock(MutationBatch.class);
-
- ArgumentCaptor<MvccEntity> writtenEntityArg = ArgumentCaptor.forClass( MvccEntity.class );
-
- when(serializationStrategy.write( same(collection), writtenEntityArg.capture() )).thenReturn( batch );
-
- Load load = new Load( uuidService, serializationStrategy, serializationFig );
- Entity loaded = load.call( entityIoEvent );
-
- assertNotNull( loaded.getField( "derp" ) );
- assertNull( loaded.getField( "derple" ) );
-
- /**
- * Verify that the repair executed
- */
- verify(batch).execute();
-
- /**
- * Verify the fields in the argument are correct
- */
- MvccEntity writtenEntity = writtenEntityArg.getValue();
-
- //assert all fields that should be updated
-
- assertSame( loaded, writtenEntity.getEntity().get() );
- assertNotNull( writtenEntity );
- assertNotNull( writtenEntity.getEntity().get().getField( "derp" ) );
- assertNull( writtenEntity.getEntity().get().getField( "derple" ) );
-
- }
-
- /**
- * Handles second trigger condition with partial updates.
- * A read on an entity , and we recognize that the entity we are reading is partial.
- */
- @Test
- public void testLoadWithPartialDelete() throws ConnectionException {
- final CollectionScope collection = mock(CollectionScope.class);
- final UUIDService uuidService = mock(UUIDService.class);
- final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
- final SerializationFig serializationFig = mock(SerializationFig.class);
- when(serializationFig.getBufferSize()).thenReturn( 10 );
-
- final ChangeLogGenerator changeLogGenerator = new ChangeLogGeneratorImpl();
-
-
-
- final UUID loadVersion = UUIDGenerator.newTimeUUID();
-
- //mock up the time uuid
- when(uuidService.newTimeUUID()).thenReturn(loadVersion);
-
- final Id entityId = TestEntityGenerator.generateId();
-
- final CollectionIoEvent<Id> entityIoEvent = new CollectionIoEvent<Id>(collection, entityId );
-
-
- final Entity entity = TestEntityGenerator.generateEntity(entityId, loadVersion);
- entity.setField( new StringField( "derp","noderp" ) );
- entity.setField( new StringField( "derple","somemerple" ) );
-
- final MvccEntity completeMvccEntity = TestEntityGenerator.fromEntityStatus( entity, MvccEntity.Status.COMPLETE );
-
-
- final Entity entity2 = TestEntityGenerator.generateEntity( entityId, UUIDGenerator.newTimeUUID() );
- entity2.setField( new StringField( "derple","somemerple" ) );
-
-
- final MvccEntity partialMvccEntity = TestEntityGenerator.fromEntityStatus( entity2, MvccEntity.Status.PARTIAL );
-
- final List<MvccEntity> results = Lists.newArrayList( partialMvccEntity );
- results.add( completeMvccEntity );
-
- //mock up returning a list of MvccEntities
- when( serializationStrategy.load( collection, entityId, loadVersion, 1 ) ).thenReturn( results.iterator());
-
-
- /**
- * Mock up returning the batch for repair
- */
-
- final MutationBatch batch = mock(MutationBatch.class);
-
- ArgumentCaptor<MvccEntity> writtenEntityArg = ArgumentCaptor.forClass( MvccEntity.class );
-
- when(serializationStrategy.write( same(collection), writtenEntityArg.capture() )).thenReturn( batch );
-
- Load load = new Load( uuidService, serializationStrategy, serializationFig );
- Entity loaded = load.call( entityIoEvent );
-
- assertNotNull( loaded.getField( "derple" ) );
-
-
- /**
- * Verify that the repair executed
- */
- verify(batch).execute();
-
- /**
- * Verify the fields in the argument are correct
- */
- MvccEntity writtenEntity = writtenEntityArg.getValue();
-
- //assert all fields that should be updated
-
- assertSame( loaded, writtenEntity.getEntity().get() );
- assertNotNull( writtenEntity );
- assertNotNull( writtenEntity.getEntity().get().getField( "derple" ) );
-
-
- }
-
- @Test
- public void testLoadWithPartialWriteDeleteThreeTimes() throws ConnectionException {
- final CollectionScope collection = mock(CollectionScope.class);
- final UUIDService uuidService = mock(UUIDService.class);
- final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
- final SerializationFig serializationFig = mock(SerializationFig.class);
- when(serializationFig.getBufferSize()).thenReturn( 10 );
-
-
- final UUID loadVersion = UUIDGenerator.newTimeUUID();
-
- //mock up the time uuid
- when(uuidService.newTimeUUID()).thenReturn(loadVersion);
-
- final Id entityId = TestEntityGenerator.generateId();
-
- final CollectionIoEvent<Id> entityIoEvent = new CollectionIoEvent<Id>(collection, entityId );
-
-
- final Entity entity = TestEntityGenerator.generateEntity(entityId, loadVersion);
- entity.setField( new StringField( "derp","noderp" ) );
- entity.setField( new StringField( "derple","somemerple" ) );
-
- final MvccEntity completeMvccEntity = TestEntityGenerator.fromEntityStatus( entity, MvccEntity.Status.COMPLETE );
-
-
- final Entity entity2 = TestEntityGenerator.generateEntity( entityId, UUIDGenerator.newTimeUUID() );
- entity2.setField( new StringField( "derple","somemerple" ) );
-
-
- final MvccEntity partialMvccEntity = TestEntityGenerator.fromEntityStatus( entity2, MvccEntity.Status.PARTIAL );
-
- final Entity entity3 = TestEntityGenerator.generateEntity( entityId, UUIDGenerator.newTimeUUID() );
- entity3.setField( new StringField( "derp","noderp" ) );
-
-
- final MvccEntity partialMvccEntity2 = TestEntityGenerator.fromEntityStatus( entity3, MvccEntity.Status.PARTIAL );
-
- final List<MvccEntity> results = Lists.newArrayList( partialMvccEntity2 );
- results.add( partialMvccEntity );
- results.add( completeMvccEntity );
-
- //mock up returning a list of MvccEntities
- when( serializationStrategy.load( collection, entityId, loadVersion, 1 ) ).thenReturn( results.iterator());
-
-
- /**
- * Mock up returning the batch for repair
- */
-
- final MutationBatch batch = mock(MutationBatch.class);
-
- ArgumentCaptor<MvccEntity> writtenEntityArg = ArgumentCaptor.forClass( MvccEntity.class );
-
- when(serializationStrategy.write( same(collection), writtenEntityArg.capture() )).thenReturn( batch );
-
- Load load = new Load( uuidService, serializationStrategy, serializationFig );
- Entity loaded = load.call( entityIoEvent );
-
- assertNotNull( loaded.getField( "derp" ) );
- assertNull( loaded.getField( "derple" ) );
-
- /**
- * Verify that the repair executed
- */
- verify(batch).execute();
-
- /**
- * Verify the fields in the argument are correct
- */
- MvccEntity writtenEntity = writtenEntityArg.getValue();
-
- //assert all fields that should be updated
-
- assertSame( loaded, writtenEntity.getEntity().get() );
- assertNotNull( writtenEntity );
- assertNotNull( writtenEntity.getEntity().get().getField( "derp" ) );
- assertNull( writtenEntity.getEntity().get().getField( "derple" ) );
-
-
-
-
- }
-
- @Test
- public void testLoadCleared(){
- final CollectionScope collection = mock(CollectionScope.class);
- final UUIDService uuidService = mock(UUIDService.class);
- final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
- final SerializationFig serializationFig = mock(SerializationFig.class);
- when(serializationFig.getBufferSize()).thenReturn( 10 );
-
-
-
- final UUID loadVersion = UUIDGenerator.newTimeUUID();
-
- //mock up the time uuid
- when(uuidService.newTimeUUID()).thenReturn(loadVersion);
-
- final Id entityId = TestEntityGenerator.generateId();
-
- final CollectionIoEvent<Id> entityIoEvent = new CollectionIoEvent<Id>(collection, entityId );
-
-
- final Iterator<MvccEntity> results = Collections.EMPTY_LIST.iterator();
-
- //mock up returning a list of MvccEntities
- when(serializationStrategy.load( collection, entityId, loadVersion, 1 )).thenReturn( results);
-
- Load load = new Load( uuidService, serializationStrategy, serializationFig );
- Entity loaded = load.call( entityIoEvent );
-
- assertNull( "No entity was loaded", loaded );
- }
-
-
-
-
-
- @Override
- protected void validateStage( final CollectionIoEvent<Id> event ) {
- final UUIDService uuidService = mock(UUIDService.class);
- final SerializationFig serializationFig = mock(SerializationFig.class);
- when(serializationFig.getBufferSize()).thenReturn( 10 );
-
-
- /**
- * Mock up wrong UUID type
- */
- when(uuidService.newTimeUUID()).thenReturn( UUID.randomUUID() );
-
-
- final MvccEntitySerializationStrategy serializationStrategy = mock(MvccEntitySerializationStrategy.class);
-
- new Load(uuidService, serializationStrategy, serializationFig).call( event );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
index f4a75d4..d1b2961 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/EntityRepairImplTest.java
@@ -38,8 +38,12 @@ import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.persistence.model.field.StringField;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+import com.netflix.astyanax.MutationBatch;
+import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
+
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -53,7 +57,7 @@ public class EntityRepairImplTest {
* Tests changing from a full version to 2 updates, ensures we have a proper output
*/
@Test
- public void testSimpleRolling() {
+ public void testSimpleRolling() throws ConnectionException {
final SerializationFig serializationFig = mock( SerializationFig.class );
@@ -101,6 +105,11 @@ public class EntityRepairImplTest {
.thenReturn( Arrays.<MvccEntity>asList( v3, v2, v1 ).iterator() );
+ final MutationBatch mutationBatch = mock( MutationBatch.class);
+
+ when(mvccEntitySerializationStrategy.write( scope, v3 )).thenReturn( mutationBatch );
+
+
EntityRepairImpl entityRepair = new EntityRepairImpl( mvccEntitySerializationStrategy, serializationFig );
final MvccEntity returned = entityRepair.maybeRepair( scope, v3 );
@@ -118,12 +127,12 @@ public class EntityRepairImplTest {
final Entity finalVersion = returned.getEntity().get();
-
- final Object expectedField1Value = v2.getEntity().get().getField( "field1" ).getValue();
-
- final Object returnedField1Value = finalVersion.getField( "field1" ).getValue();
-
- assertEquals( "Same field value", expectedField1Value, returnedField1Value );
+//
+// final Object expectedField1Value = v2.getEntity().get().getField( "field1" ).getValue();
+//
+// final Object returnedField1Value = finalVersion.getField( "field1" ).getValue();
+//
+// assertEquals( "Same field value", expectedField1Value, returnedField1Value );
@@ -132,5 +141,7 @@ public class EntityRepairImplTest {
final Object returnedField2Value = finalVersion.getField( "field2" ).getValue();
assertEquals( "Same field value", expectedField2Value, returnedField2Value );
+
+ verify(mutationBatch).execute();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f1a88aa9/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
index ba5ebce..8965668 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImplTest.java
@@ -163,7 +163,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now load it back
- MvccEntity returned = serializationStrategy.load( context, id, version );
+ MvccEntity returned = serializationStrategy.load( context, Collections.singleton( id), version ).getEntity( id );
assertEquals( "Mvcc entities are the same", saved, returned );
@@ -223,7 +223,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now get it, should be gone
- returned = serializationStrategy.load( context, id, version );
+ returned = serializationStrategy.load( context, Collections.singleton( id), version ).getEntity( id );
assertNull( returned );
}
@@ -255,7 +255,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now load it back
- MvccEntity returned = serializationStrategy.load( context, entityId, version );
+ MvccEntity returned = serializationStrategy.load( context, Collections.singleton( entityId ), version ).getEntity( entityId );
assertEquals( "Mvcc entities are the same", saved, returned );
@@ -270,7 +270,7 @@ public class MvccEntitySerializationStrategyImplTest {
serializationStrategy.mark( context, entityId, version ).execute();
- returned = serializationStrategy.load( context, entityId, version );
+ returned = serializationStrategy.load( context, Collections.singleton( entityId ), version ).getEntity( entityId );
assertEquals( entityId, returned.getId() );
assertEquals( version, returned.getVersion() );
@@ -281,7 +281,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now get it, should be gone
- returned = serializationStrategy.load( context, entityId, version );
+ returned = serializationStrategy.load( context, Collections.singleton( entityId ), version ).getEntity( entityId );
assertNull( returned );
}
@@ -318,7 +318,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now load it back
- MvccEntity returned = serializationStrategy.load( context, id, version );
+ MvccEntity returned = serializationStrategy.load( context, Collections.singleton( id ), version ).getEntity( id );
assertEquals( "Mvcc entities are the same", saved, returned );
@@ -349,7 +349,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now get it, should be gone
- returned = serializationStrategy.load( context, id, version );
+ returned = serializationStrategy.load( context, Collections.singleton( id ) , version ).getEntity( id );
assertNull( returned );
}
@@ -383,7 +383,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now load it back
- MvccEntity returnedV1 = serializationStrategy.load( context, id, version1 );
+ MvccEntity returnedV1 = serializationStrategy.load( context, Collections.singleton( id ) , version1 ).getEntity( id );
assertEquals( "Mvcc entities are the same", saved, returnedV1 );
@@ -403,7 +403,7 @@ public class MvccEntitySerializationStrategyImplTest {
serializationStrategy.write( context, savedV2 ).execute();
- MvccEntity returnedV2 = serializationStrategy.load( context, id, version2 );
+ MvccEntity returnedV2 = serializationStrategy.load( context, Collections.singleton( id ) , version2 ).getEntity( id );
assertEquals( "Mvcc entities are the same", savedV2, returnedV2 );
@@ -412,14 +412,14 @@ public class MvccEntitySerializationStrategyImplTest {
UUID version3 = UUIDGenerator.newTimeUUID();
- serializationStrategy.mark( context, id, version3 ).execute();
+ serializationStrategy.mark( context, id , version3 ).execute();
final Optional<Entity> empty = Optional.absent();
MvccEntity clearedV3 = new MvccEntityImpl( id, version3, MvccEntity.Status.COMPLETE, empty );
- MvccEntity returnedV3 = serializationStrategy.load( context, id, version3 );
+ MvccEntity returnedV3 = serializationStrategy.load( context, Collections.singleton( id ) , version3 ).getEntity( id );
assertEquals( "entities are the same", clearedV3, returnedV3 );
@@ -443,8 +443,8 @@ public class MvccEntitySerializationStrategyImplTest {
//now delete v2 and v1, we should still get v3
- serializationStrategy.delete( context, id, version1 ).execute();
- serializationStrategy.delete( context, id, version2 ).execute();
+ serializationStrategy.delete( context, id , version1 ).execute();
+ serializationStrategy.delete( context, id , version2 ).execute();
entities = serializationStrategy.load( context, id, current, 3 );
@@ -453,7 +453,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now get it, should be gone
- serializationStrategy.delete( context, id, version3 ).execute();
+ serializationStrategy.delete( context, id , version3 ).execute();
entities = serializationStrategy.load( context, id, current, 3 );
@@ -540,7 +540,7 @@ public class MvccEntitySerializationStrategyImplTest {
//now load it back
- MvccEntity returned = serializationStrategy.load( context, id, version );
+ MvccEntity returned = serializationStrategy.load( context, Collections.singleton( id ) , version ).getEntity( id );
assertEquals( "Mvcc entities are the same", saved, returned );
@@ -592,11 +592,11 @@ public class MvccEntitySerializationStrategyImplTest {
//now delete it
- serializationStrategy.delete( context, id, version ).execute();
+ serializationStrategy.delete( context, id , version ).execute();
//now get it, should be gone
- returned = serializationStrategy.load( context, id, version );
+ returned = serializationStrategy.load( context, Collections.singleton( id ) , version ).getEntity( id );
assertNull( returned );
}
@@ -641,7 +641,7 @@ public class MvccEntitySerializationStrategyImplTest {
@Test(expected = NullPointerException.class)
public void loadParamContext() throws ConnectionException {
- serializationStrategy.load( null, new SimpleId( "test" ), UUIDGenerator.newTimeUUID() );
+ serializationStrategy.load( null, Collections.<Id>emptyList(), UUIDGenerator.newTimeUUID() );
}
@@ -649,7 +649,7 @@ public class MvccEntitySerializationStrategyImplTest {
public void loadParamEntityId() throws ConnectionException {
serializationStrategy
- .load( new CollectionScopeImpl(new SimpleId( "organization" ), new SimpleId( "test" ), "test" ), (Id)null, UUIDGenerator.newTimeUUID() );
+ .load( new CollectionScopeImpl(new SimpleId( "organization" ), new SimpleId( "test" ), "test" ), null, UUIDGenerator.newTimeUUID() );
}
@@ -657,7 +657,7 @@ public class MvccEntitySerializationStrategyImplTest {
public void loadParamVersion() throws ConnectionException {
serializationStrategy
- .load( new CollectionScopeImpl(new SimpleId( "organization" ), new SimpleId( "test" ), "test" ), new SimpleId( "test" ), null );
+ .load( new CollectionScopeImpl(new SimpleId( "organization" ), new SimpleId( "test" ), "test" ), Collections.<Id>singleton( new SimpleId( "test" )), null );
}
[2/3] git commit: Changes to repair to only bring us up to the
current version only. This is what we need in practice.
Posted by to...@apache.org.
Changes to repair to only bring us up to the current version only. This is what we need in practice.
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/f7d406e3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/f7d406e3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/f7d406e3
Branch: refs/heads/collection_multiget
Commit: f7d406e3a0387b1cc5738428f49c0bd89670a91f
Parents: 185ecef
Author: Todd Nine <to...@apache.org>
Authored: Wed Oct 8 15:14:01 2014 -0600
Committer: Todd Nine <to...@apache.org>
Committed: Thu Oct 9 11:28:45 2014 -0600
----------------------------------------------------------------------
.../collection/EntityCollectionManager.java | 9 +
.../persistence/collection/EntitySet.java | 48 +++
.../impl/EntityCollectionManagerImpl.java | 10 +
.../mvcc/MvccEntitySerializationStrategy.java | 11 +
.../collection/mvcc/changelog/ChangeLog.java | 68 ++++
.../mvcc/changelog/ChangeLogEntry.java | 104 ------
.../mvcc/changelog/ChangeLogGenerator.java | 11 +-
.../mvcc/changelog/ChangeLogGeneratorImpl.java | 136 +++----
.../mvcc/changelog/ChangeLogImpl.java | 135 +++++++
.../collection/mvcc/stage/load/Load.java | 21 +-
.../collection/serialization/EntityRepair.java | 38 ++
.../serialization/impl/EntityRepairImpl.java | 146 ++++++++
.../serialization/impl/EntitySetImpl.java | 69 ++++
.../MvccEntitySerializationStrategyImpl.java | 169 ++++++---
.../persistence/collection/util/RepairUtil.java | 139 -------
.../changelog/ChangeLogGeneratorImplTest.java | 370 +++++++++++--------
.../collection/mvcc/stage/load/LoadTest.java | 30 +-
.../serialization/EntityRepairImplTest.java | 136 +++++++
...MvccEntitySerializationStrategyImplTest.java | 2 +-
19 files changed, 1121 insertions(+), 531 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
index b49989f..e5917b0 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntityCollectionManager.java
@@ -19,6 +19,8 @@
package org.apache.usergrid.persistence.collection;
+import java.util.Collection;
+
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -53,6 +55,13 @@ public interface EntityCollectionManager {
//TODO TN Change load to use multiget and return multiple entities. Only supports loading 1k per load operation.
+ /**
+ * Load all the entityIds into the observable entity set
+ * @param entityIds
+ * @return
+ */
+ public Observable<EntitySet> load(Collection<Id> entityIds);
+
//TODO Dave add a load versions using a multiget that will return a latest version structure for a collection of entity Ids
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
new file mode 100644
index 0000000..32c168f
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/EntitySet.java
@@ -0,0 +1,48 @@
+package org.apache.usergrid.persistence.collection;/*
+ * 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.
+ */
+
+
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+
+/**
+ * Represents a set of entities
+ */
+public interface EntitySet {
+
+ /**
+ * Get the entity from the result set
+ * @param entityId
+ * @return
+ */
+ public Entity getEntity(Id entityId);
+
+ /**
+ * Get the number of entities in this set
+ * @return
+ */
+ public int size();
+
+ /**
+ * Return true if the set is empty
+ * @return
+ */
+ public boolean isEmpty();
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
index b2b07e9..4a72285 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/impl/EntityCollectionManagerImpl.java
@@ -19,11 +19,14 @@
package org.apache.usergrid.persistence.collection.impl;
+import java.util.Collection;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.EntityCollectionManager;
+import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.guice.CollectionTaskExecutor;
import org.apache.usergrid.persistence.collection.guice.Write;
import org.apache.usergrid.persistence.collection.guice.WriteUpdate;
@@ -184,6 +187,13 @@ public class EntityCollectionManagerImpl implements EntityCollectionManager {
.map( load );
}
+
+ @Override
+ public Observable<EntitySet> load( final Collection<Id> entityIds ) {
+ return null;
+ }
+
+
@Override
public Observable<Entity> update( final Entity entity ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
index b2deb1a..7947029 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/MvccEntitySerializationStrategy.java
@@ -19,11 +19,13 @@
package org.apache.usergrid.persistence.collection.mvcc;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -57,6 +59,15 @@ public interface MvccEntitySerializationStrategy {
*/
public MvccEntity load( CollectionScope context, Id entityId, UUID version );
+
+ /**
+ * Load the entities into the entitySet from the specified Ids. Loads versions <= the maxVersion
+ * @param scope
+ * @param entityIds
+ * @return
+ */
+ public EntitySet load( CollectionScope scope, Collection<Id> entityIds, UUID maxVersion);
+
/**
* Load a list, from highest to lowest of the entity with versions <= version up to maxSize elements
*
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLog.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLog.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLog.java
new file mode 100644
index 0000000..d028aa0
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLog.java
@@ -0,0 +1,68 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. 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. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+
+package org.apache.usergrid.persistence.collection.mvcc.changelog;/*
+ * 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.
+ */
+
+
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+
+import org.apache.usergrid.persistence.model.field.Field;
+
+
+public interface ChangeLog {
+
+
+ /**
+ * Get all fields that should be removed from the max version
+ * @return All entries that are deletes
+ */
+ public Set<String> getDeletes();
+
+ /**
+ * Get all writes to apply to the current version
+ * @return
+ */
+ public java.util.Collection<Field> getWrites();
+
+ /**
+ * Get the number of changelog entries (writes+deletes)
+ * @return
+ */
+ public int getSize();
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogEntry.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogEntry.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogEntry.java
deleted file mode 100644
index e4b2c2b..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogEntry.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. 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. For additional information regarding
- * copyright in this work, please see the NOTICE file in the top level
- * directory of this distribution.
- */
-package org.apache.usergrid.persistence.collection.mvcc.changelog;
-
-
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.UUID;
-
-import org.apache.usergrid.persistence.model.entity.Id;
-import org.apache.usergrid.persistence.model.field.Field;
-
-import com.fasterxml.uuid.UUIDComparator;
-
-
-/**
- * Records one change to an entry field: entry ID, version, change type and the changed field.
- */
-public class ChangeLogEntry implements Comparable {
-
- private final Id entityId;
-
- private final Set<UUID> versions = new TreeSet<UUID>();
-
- public enum ChangeType {
- PROPERTY_WRITE,
- PROPERTY_DELETE
- };
-
- private final ChangeType changeType;
-
- private final Field field;
-
- public ChangeLogEntry(Id entryId, UUID version, ChangeType changeType, Field changedField) {
- this.entityId = entryId;
- if (version != null) {
- this.versions.add(version);
- }
- this.changeType = changeType;
- this.field = changedField;
- }
-
- public int compareTo( Object o ) {
- ChangeLogEntry other = (ChangeLogEntry)o;
- return UUIDComparator.staticCompare(entityId.getUuid(), other.entityId.getUuid());
- }
-
- /**
- * @return the entityId
- */
- public Id getEntryId() {
- return entityId;
- }
-
- /**
- * @return the version
- */
- public Set<UUID> getVersions() {
- return versions;
- }
-
- /**
- * @param version the version to set
- */
- public void addVersion( UUID version ) {
- this.versions.add(version);
- }
-
- /**
- * @return the changeType
- */
- public ChangeType getChangeType() {
- return changeType;
- }
-
- /**
- * @return the field
- */
- public Field getField() {
- return field;
- }
-
- public String toString() {
- return "Type = " + changeType.toString()
- + ", Property = " + field.getName()
- + ", Value = " + field.getValue()
- + ", Versions = " + versions.toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGenerator.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGenerator.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGenerator.java
index 83ff95c..68cbd58 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGenerator.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGenerator.java
@@ -18,6 +18,7 @@
package org.apache.usergrid.persistence.collection.mvcc.changelog;
+import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
@@ -38,12 +39,8 @@ public interface ChangeLogGenerator {
*
* @param versions Versions of the entity to be considered.
* Must be ordered from lowest time UUID to highest time UUID.
- *
- * @param minVersion Properties of versions older than this should be discarded
- * and properties newer should be retained.
- *
- * @return Change-log entries ordered by version, ordered from lowest time
- * uuid to highest timeuuid.
+ *
+ * @return A ChangeLog of all changes groups by version
*/
- List<ChangeLogEntry> getChangeLog( Iterator<MvccEntity> versions, UUID minVersion );
+ public ChangeLog getChangeLog( Collection<MvccEntity> versions);
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImpl.java
index 0372d3f..014d249 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImpl.java
@@ -18,13 +18,19 @@
package org.apache.usergrid.persistence.collection.mvcc.changelog;
-import java.util.*;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.field.Field;
-import com.fasterxml.uuid.UUIDComparator;
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
/**
@@ -33,79 +39,77 @@ import com.fasterxml.uuid.UUIDComparator;
public class ChangeLogGeneratorImpl implements ChangeLogGenerator {
/**
- * See parent comment
- * {@link ChangeLogGenerator#getChangeLog(java.util.Iterator<org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity>, java.util.UUID)}
- * @param mvccEntities
+ * See parent comment {@link ChangeLogGenerator#getChangeLog(java.util.Collection)}
*/
@Override
- public List<ChangeLogEntry> getChangeLog( Iterator<MvccEntity> mvccEntities, UUID minVersion ) {
+ public ChangeLog getChangeLog( Collection<MvccEntity> mvccEntities ) {
- Map<String, ChangeLogEntry> writeMap = new HashMap<String, ChangeLogEntry>();
- Map<String, ChangeLogEntry> deleteMap = new HashMap<String, ChangeLogEntry>();
- List<ChangeLogEntry> changeLog = new ArrayList<ChangeLogEntry>();
- Entity keeper = null;
- List<Entity> entityList = new ArrayList<>();
- while(mvccEntities.hasNext()) {
- MvccEntity mvccEntity = mvccEntities.next();
+ Preconditions.checkArgument( mvccEntities.size() > 0, "You must specify at least 1 entities for a change log" );
- Entity entity = mvccEntity.getEntity().get();
- entityList.add(entity);
- int compare = UUIDComparator.staticCompare(mvccEntity.getVersion(), minVersion);
+ //TODO, this is a SWAG on the entity size, this may be too little or too much.
+ final ChangeLogImpl changeLog = new ChangeLogImpl( 50 );
- if (compare == 0) {
- keeper = entity;
+ Iterator<MvccEntity> iterator = mvccEntities.iterator();
+
+ Set<String> previousFieldNames = getFieldNames( iterator.next().getEntity() );
+
+ Set<String> currentFieldNames = null;
+
+ while ( iterator.hasNext() ) {
+
+
+ final MvccEntity mvccEntity = iterator.next();
+
+ currentFieldNames = getFieldNames( mvccEntity.getEntity() );
+
+ if(mvccEntity.getStatus() == MvccEntity.Status.DELETED){
+ changeLog.clear();
+ continue;
}
- }
- for (Entity entity : entityList) {
-
- int compare = UUIDComparator.staticCompare(entity.getVersion(), minVersion);
-
-
- // TODO: what about cleared entities, all fields deleted but entity still there.
- // i.e. the optional entity will be delete
- if (compare < 0) { // less than minVersion
-
- for (Field field : entity.getFields()) {
-
- // only delete field if it is not in the keeper
- Field keeperField = keeper.getField(field.getName());
- if (keeperField == null
- || keeperField.getValue() == null
- || !keeperField.getValue().equals(field.getValue())) {
-
- String key = field.getName() + field.getValue();
- ChangeLogEntry cle = deleteMap.get(key);
- if (cle == null) {
- cle = new ChangeLogEntry(
- entity.getId(), entity.getVersion(),
- ChangeLogEntry.ChangeType.PROPERTY_DELETE, field);
- changeLog.add(cle);
- } else {
- cle.addVersion(entity.getVersion());
- }
- }
- }
-
- } else { // greater than or equal to minVersion
-
- for (Field field : entity.getFields()) {
-
- String key = field.getName() + field.getValue();
- ChangeLogEntry cle = writeMap.get(key);
- if (cle == null) {
- cle = new ChangeLogEntry(
- entity.getId(), entity.getVersion(),
- ChangeLogEntry.ChangeType.PROPERTY_WRITE, field);
- writeMap.put(key, cle);
- changeLog.add(cle);
- } else {
- cle.addVersion(entity.getVersion());
- }
- }
+ final Entity currentEntity = mvccEntity.getEntity().orNull();
+
+ //get all fields in the current field that aren't in the previous fields
+ final Set<String> deletedFields = Sets.difference( previousFieldNames, currentFieldNames );
+
+ changeLog.addDeletes( deletedFields );
+
+ for ( String addedField : currentFieldNames ) {
+ changeLog.addWrite( currentEntity.getField( addedField ) );
}
+
+
+ previousFieldNames = currentFieldNames;
+ }
+
+ //subtract off the the last set of fields from the entity
+ if(currentFieldNames != null) {
+ changeLog.clear( currentFieldNames );
}
+
+
return changeLog;
}
- }
+
+
+ /**
+ * Get all the fieldNames on this entity
+ */
+ private Set<String> getFieldNames( final Optional<Entity> entity ) {
+ if ( !entity.isPresent() ) {
+ return Collections.emptySet();
+ }
+
+
+ Collection<Field> fields = entity.get().getFields();
+
+ Set<String> fieldNames = new HashSet<>( fields.size() );
+
+ for ( final Field field : entity.get().getFields() ) {
+ fieldNames.add( field.getName() );
+ }
+
+ return fieldNames;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogImpl.java
new file mode 100644
index 0000000..e931a1f
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogImpl.java
@@ -0,0 +1,135 @@
+/*
+ *
+ * * Licensed to the Apache Software Foundation (ASF) under one or more
+ * * contributor license agreements. 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. For additional information regarding
+ * * copyright in this work, please see the NOTICE file in the top level
+ * * directory of this distribution.
+ *
+ */
+
+package org.apache.usergrid.persistence.collection.mvcc.changelog;/*
+ * 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.
+ */
+
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.usergrid.persistence.model.field.Field;
+
+
+/**
+ * Create a new instance of the change log
+ */
+public class ChangeLogImpl implements ChangeLog {
+
+ private final HashMap<String, Field> additions;
+ private final HashSet<String> deletions;
+
+
+ /**
+ * @param approximateMaxFieldCount The estimated upper bounds of the changelog, helps with efficiency by
+ * overallocation if necessary
+ */
+ public ChangeLogImpl( final int approximateMaxFieldCount ) {
+ //todo, may be way too large
+ additions = new HashMap( approximateMaxFieldCount );
+ deletions = new HashSet( approximateMaxFieldCount );
+ }
+
+
+ /**
+ * Add change logs to this log
+ *
+ * @param deletes The set containing all string fields to delete
+ */
+ public void addDeletes( final Set<String> deletes ) {
+ this.deletions.addAll( deletes );
+ clearAdditions( deletes );
+ }
+
+
+ /**
+ * Add the field as a write column
+ */
+ public void addWrite( final Field field ) {
+ final String fieldName = field.getName();
+ this.additions.put( field.getName(), field );
+ this.deletions.remove( fieldName );
+ }
+
+
+ /**
+ * Remove the names from the additions and deletions
+ */
+ public void clear( final Set<String> names ) {
+ this.deletions.removeAll( names );
+ clearAdditions( names );
+ }
+
+
+ /**
+ * Clear additions by name
+ */
+ private void clearAdditions( final Set<String> deletes ) {
+
+ for ( final String key : deletes ) {
+ this.additions.remove( key );
+ }
+ }
+
+
+ /**
+ * Clear all all additions and deletions
+ */
+ public void clear() {
+ this.additions.clear();
+ this.deletions.clear();
+ }
+
+
+ @Override
+ public Set<String> getDeletes() {
+ return deletions;
+ }
+
+
+ @Override
+ public Collection<Field> getWrites() {
+ return additions.values();
+ }
+
+
+ @Override
+ public int getSize() {
+ return deletions.size() + additions.size();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
index 9b7f437..f9c66cc 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/mvcc/stage/load/Load.java
@@ -29,8 +29,10 @@ import org.apache.usergrid.persistence.collection.CollectionScope;
import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.collection.mvcc.stage.CollectionIoEvent;
+import org.apache.usergrid.persistence.collection.serialization.EntityRepair;
+import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
+import org.apache.usergrid.persistence.collection.serialization.impl.EntityRepairImpl;
import org.apache.usergrid.persistence.collection.service.UUIDService;
-import org.apache.usergrid.persistence.collection.util.RepairUtil;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -53,16 +55,20 @@ public class Load implements Func1<CollectionIoEvent<Id>, Entity> {
private final UUIDService uuidService;
private final MvccEntitySerializationStrategy entitySerializationStrategy;
+ private final EntityRepair entityRepair;
@Inject
- public Load( final UUIDService uuidService, final MvccEntitySerializationStrategy entitySerializationStrategy ) {
+ public Load( final UUIDService uuidService, final MvccEntitySerializationStrategy entitySerializationStrategy, final
+ SerializationFig serializationFig ) {
Preconditions.checkNotNull( entitySerializationStrategy, "entitySerializationStrategy is required" );
Preconditions.checkNotNull( uuidService, "uuidService is required" );
this.uuidService = uuidService;
this.entitySerializationStrategy = entitySerializationStrategy;
+ entityRepair = new EntityRepairImpl( entitySerializationStrategy, serializationFig );
+
}
//TODO: do reads partial merges in batches. maybe 5 or 10 at a time.
@@ -88,11 +94,18 @@ public class Load implements Func1<CollectionIoEvent<Id>, Entity> {
Iterator<MvccEntity> results = entitySerializationStrategy.load(
collectionScope, entityId, versionMax, 1 );
- MvccEntity repairedEntity = RepairUtil.repair( results,collectionScope,entitySerializationStrategy );
+ if(!results.hasNext()){
+ return null;
+ }
+
+ final MvccEntity returned = results.next();
+
+ final MvccEntity repairedEntity = entityRepair.maybeRepair( collectionScope, returned );
+
if(repairedEntity == null)
return null;
- return repairedEntity.getEntity().get();
+ return repairedEntity.getEntity().orNull();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/EntityRepair.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/EntityRepair.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/EntityRepair.java
new file mode 100644
index 0000000..c4fddd2
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/EntityRepair.java
@@ -0,0 +1,38 @@
+package org.apache.usergrid.persistence.collection.serialization;/*
+ * 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.
+ */
+
+
+import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
+
+
+/**
+ * Interface for entity repair operations
+ */
+public interface EntityRepair {
+
+ /**
+ * Run the repair task for this entity. If the entity does not need repaired, it will just be returned
+ *
+ * @param collectionScope The scope of the entity to possibly repair
+ * @param targetEntity The entity to check and repair
+ * @return The source entity or the repaired entity
+ */
+ public MvccEntity maybeRepair(CollectionScope collectionScope, MvccEntity targetEntity);
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
new file mode 100644
index 0000000..d26cd3f
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntityRepairImpl.java
@@ -0,0 +1,146 @@
+package org.apache.usergrid.persistence.collection.serialization.impl;/*
+ * 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.
+ */
+
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
+import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLog;
+import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGenerator;
+import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGeneratorImpl;
+import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
+import org.apache.usergrid.persistence.collection.serialization.EntityRepair;
+import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.field.Field;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+
+@Singleton
+public class EntityRepairImpl implements EntityRepair {
+
+
+ private static final ChangeLogGenerator CHANGE_LOG_GENERATOR = new ChangeLogGeneratorImpl();
+
+ private final MvccEntitySerializationStrategy mvccEntitySerializationStrategy;
+ private final SerializationFig serializationFig;
+
+
+ @Inject
+ public EntityRepairImpl( final MvccEntitySerializationStrategy mvccEntitySerializationStrategy,
+ final SerializationFig serializationFig ) {
+ this.mvccEntitySerializationStrategy = mvccEntitySerializationStrategy;
+ this.serializationFig = serializationFig;
+ }
+
+
+ @Override
+ public MvccEntity maybeRepair( final CollectionScope collectionScope, final MvccEntity targetEntity ) {
+ if ( !needsRepaired( targetEntity ) ) {
+ return targetEntity;
+ }
+
+
+ final List<MvccEntity> partialEntities = new ArrayList<>( serializationFig.getBufferSize() );
+
+ partialEntities.add( targetEntity );
+
+ final Iterator<MvccEntity> results = mvccEntitySerializationStrategy
+ .load( collectionScope, targetEntity.getId(), targetEntity.getVersion(),
+ serializationFig.getBufferSize() );
+
+
+
+ //discard the one that's equal to the version we were passed in
+ if(results.hasNext() ){
+ results.next();
+ }
+
+
+// MvccEntity oldestCompleteEntity;
+
+
+ while ( results.hasNext() ) {
+ final MvccEntity mvccEntity = results.next();
+ partialEntities.add( mvccEntity );
+
+
+ if ( !needsRepaired( mvccEntity ) ) {
+ break;
+ }
+ }
+
+ Collections.reverse( partialEntities );
+
+ final ChangeLog changeLog =
+ CHANGE_LOG_GENERATOR.getChangeLog( partialEntities );
+
+
+ //repair
+ final MvccEntity mergedEntity = entityRepair( changeLog, targetEntity );
+
+ try {
+ mvccEntitySerializationStrategy.write( collectionScope, mergedEntity ).execute();
+ }
+ catch ( Exception e ) {
+ throw new RuntimeException( "Couldn't rewrite repaired entity", e );
+ }
+
+ return mergedEntity;
+
+ }
+
+
+ /**
+ * Appies changes to the entity log, oldest to newest version
+ */
+ private MvccEntity entityRepair( final ChangeLog changeLog, final MvccEntity targetEntity ) {
+
+ //TODO review this, why do we care other than just replaying the changelog?
+
+ final Entity entity = targetEntity.getEntity().get();
+
+ for(final String removedField: changeLog.getDeletes()){
+ entity.removeField( removedField );
+ }
+
+ for(final Field newField : changeLog.getWrites()){
+ entity.setField( newField );
+ }
+
+ return targetEntity;
+ }
+
+
+ /**
+ * Returns true if the entity needs repaired
+ */
+ private boolean needsRepaired( final MvccEntity headEntity ) {
+
+ final MvccEntity.Status status = headEntity.getStatus();
+
+ return !(status == MvccEntity.Status.COMPLETE || status == MvccEntity.Status.DELETED || !headEntity.getEntity().isPresent());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
new file mode 100644
index 0000000..89ec69e
--- /dev/null
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/EntitySetImpl.java
@@ -0,0 +1,69 @@
+package org.apache.usergrid.persistence.collection.serialization.impl;/*
+ * 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.
+ */
+
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.usergrid.persistence.collection.EntitySet;
+import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
+import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+
+public class EntitySetImpl implements EntitySet {
+
+
+ private final Map<Id, MvccEntity> entities;
+
+
+ public EntitySetImpl(
+ final int expectedSize ) {
+ this.entities = new HashMap<>( expectedSize );
+ }
+
+
+ public void addEntity( final MvccEntity entity ) {
+ entities.put( entity.getId(), entity );
+ }
+
+
+ @Override
+ public Entity getEntity( final Id entityId ) {
+ final MvccEntity entity = entities.get( entityId );
+
+ if ( entity == null ) {
+ return null;
+ }
+
+ return entity.getEntity().orNull();
+ }
+
+
+ @Override
+ public int size() {
+ return entities.size();
+ }
+
+
+ @Override
+ public boolean isEmpty() {
+ return entities.size() == 0;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
index 419d0f0..b36ee0e 100644
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
+++ b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/serialization/impl/MvccEntitySerializationStrategyImpl.java
@@ -19,20 +19,29 @@ package org.apache.usergrid.persistence.collection.serialization.impl;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
+import java.util.List;
import java.util.UUID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.ReversedType;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.usergrid.persistence.collection.CollectionScope;
+import org.apache.usergrid.persistence.collection.EntitySet;
import org.apache.usergrid.persistence.collection.exception.CollectionRuntimeException;
import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
import org.apache.usergrid.persistence.collection.mvcc.entity.impl.MvccEntityImpl;
+import org.apache.usergrid.persistence.collection.serialization.EntityRepair;
+import org.apache.usergrid.persistence.collection.serialization.SerializationFig;
import org.apache.usergrid.persistence.collection.util.EntityUtils;
import org.apache.usergrid.persistence.core.astyanax.ColumnNameIterator;
import org.apache.usergrid.persistence.core.astyanax.ColumnParser;
@@ -45,7 +54,6 @@ import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.dataformat.smile.SmileFactory;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
@@ -57,16 +65,17 @@ import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.connectionpool.exceptions.NotFoundException;
import com.netflix.astyanax.model.Column;
+import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.model.CompositeBuilder;
import com.netflix.astyanax.model.CompositeParser;
import com.netflix.astyanax.model.Composites;
+import com.netflix.astyanax.model.Row;
import com.netflix.astyanax.query.RowQuery;
import com.netflix.astyanax.serializers.AbstractSerializer;
import com.netflix.astyanax.serializers.ByteBufferSerializer;
import com.netflix.astyanax.serializers.BytesArraySerializer;
import com.netflix.astyanax.serializers.UUIDSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
/**
* @author tnine
@@ -74,9 +83,9 @@ import org.slf4j.LoggerFactory;
@Singleton
public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializationStrategy, Migration {
- private static final Logger log = LoggerFactory.getLogger( MvccLogEntrySerializationStrategyImpl.class );
+ private static final Logger log = LoggerFactory.getLogger( MvccLogEntrySerializationStrategyImpl.class );
- private static final EntitySerializer SER = new EntitySerializer();
+ private static final EntitySerializer ENTITY_JSON_SER = new EntitySerializer();
private static final IdRowCompositeSerializer ID_SER = IdRowCompositeSerializer.get();
@@ -94,11 +103,15 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
protected final Keyspace keyspace;
+ protected final SerializationFig serializationFig;
+ protected final EntityRepair repair;
@Inject
- public MvccEntitySerializationStrategyImpl( final Keyspace keyspace ) {
+ public MvccEntitySerializationStrategyImpl( final Keyspace keyspace, final SerializationFig serializationFig ) {
this.keyspace = keyspace;
+ this.serializationFig = serializationFig;
+ this.repair = new EntityRepairImpl( this, serializationFig );
}
@@ -114,9 +127,10 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
try {
- colMutation.putColumn( colName,
- SER.toByteBuffer( new EntityWrapper( entity.getStatus(), entity.getEntity() ) ) );
- } catch ( Exception e ) {
+ colMutation.putColumn( colName, ENTITY_JSON_SER
+ .toByteBuffer( new EntityWrapper( entity.getStatus(), entity.getEntity() ) ) );
+ }
+ catch ( Exception e ) {
// throw better exception if we can
if ( entity != null || entity.getEntity().get() != null ) {
throw new CollectionRuntimeException( entity, collectionScope, e );
@@ -147,34 +161,79 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
return null;
}
catch ( ConnectionException e ) {
- throw new CollectionRuntimeException( null, collectionScope, "An error occurred connecting to cassandra", e );
+ throw new CollectionRuntimeException( null, collectionScope, "An error occurred connecting to cassandra",
+ e );
}
- return new MvccColumnParser(entityId).parseColumn(column);
+ return new MvccColumnParser( entityId ).parseColumn( column );
}
@Override
- public Iterator<MvccEntity> load( final CollectionScope collectionScope, final Id entityId, final UUID version,
- final int fetchSize ) {
+ public EntitySet load( final CollectionScope collectionScope, final Collection<Id> entityIds,
+ final UUID maxVersion ) {
+
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
- Preconditions.checkNotNull( entityId, "entity id is required" );
- Preconditions.checkNotNull( version, "version is required" );
- Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );
+ Preconditions.checkNotNull( entityIds, "entityIds is required" );
+ Preconditions.checkArgument( entityIds.size() > 0, "entityIds is required" );
+ Preconditions.checkNotNull( maxVersion, "version is required" );
+ final List<ScopedRowKey<CollectionScope, Id>> rowKeys = new ArrayList<>( entityIds.size() );
+
+
+ for ( final Id entityId : entityIds ) {
+ rowKeys.add( ScopedRowKey.fromKey( collectionScope, entityId ) );
+ }
+
+
+ final Iterator<Row<ScopedRowKey<CollectionScope, Id>, UUID>> latestEntityColumns;
+
+
+
+ try {
+ latestEntityColumns = keyspace.prepareQuery( CF_ENTITY_DATA ).getKeySlice( rowKeys )
+ .withColumnRange( maxVersion, null, false, 1 ).execute().getResult()
+ .iterator();
+ } catch ( ConnectionException e ) {
+ throw new CollectionRuntimeException( null, collectionScope, "An error occurred connecting to cassandra",
+ e );
+ }
+
+
+
+ final EntitySetImpl entitySetResults = new EntitySetImpl( entityIds.size() );
+
+ while ( latestEntityColumns.hasNext() ) {
+ final Row<ScopedRowKey<CollectionScope, Id>, UUID> row = latestEntityColumns.next();
+
+ final ColumnList<UUID> columns = row.getColumns();
+
+ if ( columns.size() == 0 ) {
+ continue;
+ }
- RowQuery<ScopedRowKey<CollectionScope, Id>, UUID> query = keyspace.prepareQuery(CF_ENTITY_DATA).getKey(ScopedRowKey.fromKey(collectionScope, entityId))
- .withColumnRange(version, null, false, fetchSize);
+ final Id entityId = row.getKey().getKey();
- return new ColumnNameIterator(query, new MvccColumnParser(entityId), false);
+ final Column<UUID> column = columns.getColumnByIndex( 0 );
+ final MvccEntity parsedEntity = new MvccColumnParser( entityId ).parseColumn( column );
+
+ //we *might* need to repair, it's not clear so check before loading into result sets
+ final MvccEntity maybeRepaired = repair.maybeRepair( collectionScope, parsedEntity );
+
+ entitySetResults.addEntity( maybeRepaired );
+
+ }
+
+ return entitySetResults;
}
+
@Override
- public Iterator<MvccEntity> loadHistory( final CollectionScope collectionScope, final Id entityId, final UUID version,
+ public Iterator<MvccEntity> load( final CollectionScope collectionScope, final Id entityId, final UUID version,
final int fetchSize ) {
Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
@@ -182,11 +241,29 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
Preconditions.checkNotNull( version, "version is required" );
Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );
- RowQuery<ScopedRowKey<CollectionScope, Id>, UUID> query = keyspace.prepareQuery(CF_ENTITY_DATA).getKey(ScopedRowKey.fromKey(collectionScope, entityId))
- .withColumnRange(null, version, true, fetchSize);
- return new ColumnNameIterator(query, new MvccColumnParser(entityId), false);
+ RowQuery<ScopedRowKey<CollectionScope, Id>, UUID> query =
+ keyspace.prepareQuery( CF_ENTITY_DATA ).getKey( ScopedRowKey.fromKey( collectionScope, entityId ) )
+ .withColumnRange( version, null, false, fetchSize );
+
+ return new ColumnNameIterator( query, new MvccColumnParser( entityId ), false );
+ }
+
+
+ @Override
+ public Iterator<MvccEntity> loadHistory( final CollectionScope collectionScope, final Id entityId,
+ final UUID version, final int fetchSize ) {
+
+ Preconditions.checkNotNull( collectionScope, "collectionScope is required" );
+ Preconditions.checkNotNull( entityId, "entity id is required" );
+ Preconditions.checkNotNull( version, "version is required" );
+ Preconditions.checkArgument( fetchSize > 0, "max Size must be greater than 0" );
+
+ RowQuery<ScopedRowKey<CollectionScope, Id>, UUID> query =
+ keyspace.prepareQuery( CF_ENTITY_DATA ).getKey( ScopedRowKey.fromKey( collectionScope, entityId ) )
+ .withColumnRange( null, version, true, fetchSize );
+ return new ColumnNameIterator( query, new MvccColumnParser( entityId ), false );
}
@@ -201,8 +278,8 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
return doWrite( collectionScope, entityId, new RowOp() {
@Override
public void doOp( final ColumnListMutation<UUID> colMutation ) {
- colMutation.putColumn( version, SER.toByteBuffer(
- new EntityWrapper( MvccEntity.Status.COMPLETE, Optional.<Entity>absent() ) ) );
+ colMutation.putColumn( version, ENTITY_JSON_SER
+ .toByteBuffer( new EntityWrapper( MvccEntity.Status.COMPLETE, Optional.<Entity>absent() ) ) );
}
} );
}
@@ -229,8 +306,10 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
//create the CF entity data. We want it reversed b/c we want the most recent version at the top of the
//row for fast seeks
- MultiTennantColumnFamilyDefinition cf = new MultiTennantColumnFamilyDefinition( CF_ENTITY_DATA,
- BytesType.class.getSimpleName(), ReversedType.class.getSimpleName() + "(" + UUIDType.class.getSimpleName() + ")", BytesType.class.getSimpleName(), MultiTennantColumnFamilyDefinition.CacheOption.KEYS );
+ MultiTennantColumnFamilyDefinition cf =
+ new MultiTennantColumnFamilyDefinition( CF_ENTITY_DATA, BytesType.class.getSimpleName(),
+ ReversedType.class.getSimpleName() + "(" + UUIDType.class.getSimpleName() + ")",
+ BytesType.class.getSimpleName(), MultiTennantColumnFamilyDefinition.CacheOption.KEYS );
return Collections.singleton( cf );
@@ -249,7 +328,6 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
}
-
/**
* Simple callback to perform puts and deletes with a common row setup code
*/
@@ -276,23 +354,24 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
}
}
+
/**
* Converts raw columns the to MvccEntity representation
-
*/
private static final class MvccColumnParser implements ColumnParser<UUID, MvccEntity> {
private final Id id;
- private MvccColumnParser(Id id) {
+
+ private MvccColumnParser( Id id ) {
this.id = id;
}
@Override
- public MvccEntity parseColumn(Column<UUID> column) {
+ public MvccEntity parseColumn( Column<UUID> column ) {
- final EntityWrapper deSerialized = column.getValue( SER );
+ final EntityWrapper deSerialized = column.getValue( ENTITY_JSON_SER );
//Inject the id into it.
if ( deSerialized.entity.isPresent() ) {
@@ -303,11 +382,12 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
}
}
+
public static class EntitySerializer extends AbstractSerializer<EntityWrapper> {
public static final EntitySerializer INSTANCE = new EntitySerializer();
- public static final SmileFactory f = new SmileFactory( );
+ public static final SmileFactory f = new SmileFactory();
public static ObjectMapper mapper;
@@ -321,16 +401,20 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
//the marker for when we're passed a "null" value
private static final byte[] EMPTY = new byte[] { 0x0 };
+
public EntitySerializer() {
try {
mapper = new ObjectMapper( f );
-// mapper.enable(SerializationFeature.INDENT_OUTPUT); don't indent output, causes slowness
- mapper.enableDefaultTypingAsProperty(ObjectMapper.DefaultTyping.JAVA_LANG_OBJECT, "@class");
- } catch ( Exception e ) {
- throw new RuntimeException("Error setting up mapper", e);
+ // mapper.enable(SerializationFeature.INDENT_OUTPUT); don't indent output,
+ // causes slowness
+ mapper.enableDefaultTypingAsProperty( ObjectMapper.DefaultTyping.JAVA_LANG_OBJECT, "@class" );
+ }
+ catch ( Exception e ) {
+ throw new RuntimeException( "Error setting up mapper", e );
}
}
+
@Override
public ByteBuffer toByteBuffer( final EntityWrapper wrapper ) {
if ( wrapper == null ) {
@@ -363,15 +447,16 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
builder.addBytes( mapper.writeValueAsBytes( wrapper.entity.get() ) );
}
catch ( Exception e ) {
- throw new RuntimeException("Unable to serialize entity", e);
+ throw new RuntimeException( "Unable to serialize entity", e );
}
return builder.build();
}
+
@Override
public EntityWrapper fromByteBuffer( final ByteBuffer byteBuffer ) {
- CompositeParser parser = Composites.newCompositeParser( byteBuffer );
+ CompositeParser parser = Composites.newCompositeParser( byteBuffer );
byte[] version = parser.read( BYTES_ARRAY_SERIALIZER );
@@ -389,7 +474,7 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
Entity storedEntity = null;
- ByteBuffer jsonBytes = parser.read( BUFFER_SERIALIZER );
+ ByteBuffer jsonBytes = parser.read( BUFFER_SERIALIZER );
byte[] array = jsonBytes.array();
int start = jsonBytes.arrayOffset();
int length = jsonBytes.remaining();
@@ -398,10 +483,10 @@ public class MvccEntitySerializationStrategyImpl implements MvccEntitySerializat
storedEntity = mapper.readValue( array, start, length, Entity.class );
}
catch ( Exception e ) {
- throw new RuntimeException("Unable to read entity data", e);
+ throw new RuntimeException( "Unable to read entity data", e );
}
- final Optional<Entity> entity = Optional.of( storedEntity);
+ final Optional<Entity> entity = Optional.of( storedEntity );
if ( Arrays.equals( STATE_COMPLETE, state ) ) {
return new EntityWrapper( MvccEntity.Status.COMPLETE, entity );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/RepairUtil.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/RepairUtil.java b/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/RepairUtil.java
deleted file mode 100644
index aac5794..0000000
--- a/stack/corepersistence/collection/src/main/java/org/apache/usergrid/persistence/collection/util/RepairUtil.java
+++ /dev/null
@@ -1,139 +0,0 @@
-package org.apache.usergrid.persistence.collection.util;
-
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-
-import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
-import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogEntry;
-import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGenerator;
-import org.apache.usergrid.persistence.collection.mvcc.changelog.ChangeLogGeneratorImpl;
-import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
-
-
-/**
- * This class holds different methods pertaining to the consolidation of an mvccentity.
- *
- */
-public class RepairUtil {
-
- private static final ChangeLogGenerator changeLogGenerator = new ChangeLogGeneratorImpl();
-
-
- /**
- * Name of the operation to be done on the entity. Repair starts the process for determining whether an
- * entity needs repair or not.
- * @param results
- * @param collectionScope
- * @param entitySerializationStrategy
- * @return
- */
- public static MvccEntity repair( Iterator<MvccEntity> results, CollectionScope collectionScope,
- MvccEntitySerializationStrategy entitySerializationStrategy ) {
-
- //nothing to do, we didn't get a result back
- if ( !results.hasNext() ) {
- return null;
- }
-
- final MvccEntity partialEntity = results.next();
- List<MvccEntity> partialEntities = new ArrayList<>();
-
- //this entity has been marked as cleared.(deleted)
- //The version exists, but does not have entity data
- if ( !partialEntity.getEntity().isPresent() && ( partialEntity.getStatus() == MvccEntity.Status.DELETED
- || partialEntity.getStatus() == MvccEntity.Status.COMPLETE ) ) {
- return null;
- }
-
- if ( partialEntity.getStatus() == MvccEntity.Status.COMPLETE ) {
- return partialEntity;
- }
-
- partialEntities.add( partialEntity );
-
-
- while ( results.hasNext() ) {
- final MvccEntity previousEntity = results.next();
- partialEntities.add( previousEntity );
-
- if ( previousEntity.getStatus() != MvccEntity.Status.PARTIAL ) {
- return repairAndWrite( partialEntities, partialEntity, entitySerializationStrategy, collectionScope );
- }
- }
- return null;
- }
-
-
- /**
- * Repairs the entity then overwrites the previous entity to become the new completed entity.
- * @param partialEntities
- * @param targetEntity The entity that should ultimately contain all merged data
- * @param entitySerializationStrategy
- * @param collectionScope
- * @return
- */
- private static MvccEntity repairAndWrite( List<MvccEntity> partialEntities, MvccEntity targetEntity,
- MvccEntitySerializationStrategy entitySerializationStrategy,
- CollectionScope collectionScope ) {
- Collections.reverse( partialEntities );
-
- //repair
- final MvccEntity mergedEntity = entityRepair( changeLogGenerator.getChangeLog( partialEntities.iterator(),
- partialEntities.get( partialEntities.size() - 1 ).getVersion() ), partialEntities, targetEntity
- );
-
- try {
- entitySerializationStrategy.write( collectionScope, mergedEntity ).execute();
- }
- catch ( Exception e ) {
- throw new RuntimeException( "Couldn't rewrite repaired entity", e );
- }
- return mergedEntity;
- }
-
-
- /**
- * Applies the changelog to the completed entity.
- * @param changeLogEntryList
- * @param results
- * @param completedEntity
- * @return
- */
- private static MvccEntity entityRepair( List<ChangeLogEntry> changeLogEntryList, List<MvccEntity> results,
- MvccEntity completedEntity ) {
- int changeLogIndex = 0;
- for ( MvccEntity result : results ) {
-
- while ( changeLogIndex != changeLogEntryList.size() ) {
-
- ChangeLogEntry changeLogEntry = changeLogEntryList.get( changeLogIndex );
-
- if ( result.getId().equals( changeLogEntry.getEntryId() ) ) {
-
- ChangeLogEntry.ChangeType changeType = changeLogEntry.getChangeType();
-
- if ( changeType.equals( ChangeLogEntry.ChangeType.PROPERTY_DELETE ) ) {
- completedEntity.getEntity().get().getFields().remove( changeLogEntry.getField() );
- }
- else if ( changeType.equals( ChangeLogEntry.ChangeType.PROPERTY_WRITE ) ) {
- completedEntity.getEntity().get().setField( changeLogEntry.getField() );
- }
- changeLogIndex++;
- }
- else {
- break;
- }
- }
- }
-
- if ( !completedEntity.getEntity().isPresent() ) {
- return null;
- }
-
- return completedEntity;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/f7d406e3/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImplTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImplTest.java b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImplTest.java
index 56792a9..6795893 100644
--- a/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImplTest.java
+++ b/stack/corepersistence/collection/src/test/java/org/apache/usergrid/persistence/collection/mvcc/changelog/ChangeLogGeneratorImplTest.java
@@ -18,215 +18,261 @@
package org.apache.usergrid.persistence.collection.mvcc.changelog;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
-import java.util.List;
+import java.util.Set;
-import org.jukito.JukitoModule;
-import org.jukito.UseModules;
-import org.junit.Assert;
-import org.junit.ClassRule;
-import org.junit.Rule;
import org.junit.Test;
-import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.usergrid.persistence.collection.CollectionScope;
-import org.apache.usergrid.persistence.collection.EntityCollectionManager;
-import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
-import org.apache.usergrid.persistence.collection.guice.CollectionModule;
-import org.apache.usergrid.persistence.collection.guice.MigrationManagerRule;
-import org.apache.usergrid.persistence.collection.guice.TestCollectionModule;
-import org.apache.usergrid.persistence.collection.impl.CollectionScopeImpl;
-import org.apache.usergrid.persistence.collection.mvcc.MvccEntitySerializationStrategy;
import org.apache.usergrid.persistence.collection.mvcc.entity.MvccEntity;
-import org.apache.usergrid.persistence.core.cassandra.CassandraRule;
-import org.apache.usergrid.persistence.core.cassandra.ITRunner;
+import org.apache.usergrid.persistence.collection.mvcc.entity.impl.MvccEntityImpl;
import org.apache.usergrid.persistence.model.entity.Entity;
+import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
+import org.apache.usergrid.persistence.model.field.BooleanField;
+import org.apache.usergrid.persistence.model.field.Field;
import org.apache.usergrid.persistence.model.field.IntegerField;
import org.apache.usergrid.persistence.model.field.StringField;
+import org.apache.usergrid.persistence.model.util.UUIDGenerator;
-import com.google.inject.Inject;
+import com.google.common.base.Optional;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
-import rx.Observable;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
/**
* Test basic operation of change log
*/
-@RunWith( ITRunner.class )
-@UseModules( TestCollectionModule.class )
public class ChangeLogGeneratorImplTest {
private static final Logger LOG = LoggerFactory.getLogger( ChangeLogGeneratorImplTest.class );
- @Inject
- @Rule
- public MigrationManagerRule migrationManagerRule;
+ /**
+ * Test rolling up 3 versions, properties are added then deleted
+ */
+ @Test
+ public void testBasicOperation() throws ConnectionException {
+
+ LOG.info( "ChangeLogGeneratorImpl test" );
+
+
+ final Id entityId = new SimpleId( "test" );
+
+ Entity e1 = new Entity( entityId );
+ e1.setField( new StringField( "name", "name1" ) );
+ e1.setField( new IntegerField( "count", 1 ) );
+ e1.setField( new BooleanField( "single", true ) );
+
+ final MvccEntity mvccEntity1 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.COMPLETE, e1 );
+
+ Entity e2 = new Entity( entityId );
+ e2.setField( new StringField( "name", "name2" ) );
+ e2.setField( new IntegerField( "count", 2 ) );
+ e2.setField( new StringField( "nickname", "buddy" ) );
+ e2.setField( new BooleanField( "cool", false ) );
+
+ final MvccEntity mvccEntity2 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, e2 );
+
+
+ Entity e3 = new Entity( entityId );
+ e3.setField( new StringField( "name", "name3" ) );
+ e3.setField( new IntegerField( "count", 2 ) );
+ //appears in e1, since it's been added again, we want to make sure it doesn't appear in the delete list
+ e3.setField( new BooleanField( "single", true ) );
+
+ final MvccEntity mvccEntity3 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, e3 );
+
+
+ ChangeLogGeneratorImpl instance = new ChangeLogGeneratorImpl();
+ ChangeLog result =
+ instance.getChangeLog( Arrays.asList( mvccEntity1, mvccEntity2, mvccEntity3 ) ); // minVersion = e3
+
+
+ assertEquals( "All changes not present", 2, result.getSize() );
+
- @Inject
- private EntityCollectionManagerFactory factory;
+ Collection<Field> changes = result.getWrites();
+
+ assertEquals( 0, changes.size() );
+
+ Set<String> deletes = result.getDeletes();
+
+ assertEquals( 2, deletes.size() );
+
+ assertTrue( deletes.contains( "nickname" ) );
+ assertTrue( deletes.contains( "cool" ) );
+ }
- @Inject
- MvccEntitySerializationStrategy mvccEntitySerializationStrategy;
/**
- * Test that change log creation follows Todd's example.
- * TODO, can we do this without doing serialization I/O on the entities?
- * This seems out of the scope of the changelog itself
+ * Test rolling up 3 versions, properties are added then deleted
*/
@Test
- public void testBasicOperation() throws ConnectionException {
+ public void testDeletedVersionFirst() throws ConnectionException {
+
+ LOG.info( "ChangeLogGeneratorImpl test" );
+
+
+ final Id entityId = new SimpleId( "test" );
+
+ final MvccEntity mvccEntity1 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.DELETED,
+ Optional.<Entity>absent() );
+
+ Entity e2 = new Entity( entityId );
+ e2.setField( new StringField( "name", "name2" ) );
+ e2.setField( new IntegerField( "count", 2 ) );
+ e2.setField( new StringField( "nickname", "buddy" ) );
+ e2.setField( new BooleanField( "cool", false ) );
+
+ final MvccEntity mvccEntity2 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, e2 );
+
+
+ Entity e3 = new Entity( entityId );
+ e3.setField( new StringField( "name", "name3" ) );
+ e3.setField( new IntegerField( "count", 2 ) );
+ //appears in e1, since it's been added again, we want to make sure it doesn't appear in the delete list
+ e3.setField( new BooleanField( "single", true ) );
+
+ final MvccEntity mvccEntity3 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, e3 );
+
+
+ ChangeLogGeneratorImpl instance = new ChangeLogGeneratorImpl();
+ ChangeLog result =
+ instance.getChangeLog( Arrays.asList( mvccEntity1, mvccEntity2, mvccEntity3 ) ); // minVersion = e3
+
+
+ assertEquals( "All changes not present", 2, result.getSize() );
+
- LOG.info("ChangeLogGeneratorImpl test");
+ Collection<Field> changes = result.getWrites();
- // create an entity and make a series of changes to it so that versions get created
- CollectionScope context = new CollectionScopeImpl(
- new SimpleId( "organization" ), new SimpleId( "test" ), "test" );
+ assertEquals( 0, changes.size() );
- // Todd's example:
- //
- // V1 : { "name" : "name1" , "count": 1}
- // V2: { "name" : "name2" , "count": 2, "nickname" : "buddy"}
- // V3: { "name" : "name3" , "count": 2}
-
- EntityCollectionManager manager = factory.createCollectionManager( context );
- Entity e1 = new Entity( new SimpleId( "test" ) );
+
+
+ Set<String> deletes = result.getDeletes();
+
+ assertEquals( 2, deletes.size() );
+
+ assertTrue( deletes.contains( "nickname" ) );
+ assertTrue( deletes.contains( "cool" ) );
+ }
+
+
+ /**
+ * Test rolling up 3 versions, properties are added then deleted
+ */
+ @Test
+ public void testDeletedMiddle() throws ConnectionException {
+
+ LOG.info( "ChangeLogGeneratorImpl test" );
+
+
+ final Id entityId = new SimpleId( "test" );
+
+ Entity e1 = new Entity( entityId );
e1.setField( new StringField( "name", "name1" ) );
e1.setField( new IntegerField( "count", 1 ) );
- Observable<Entity> o1 = manager.write( e1 );
- e1 = o1.toBlocking().lastOrDefault( null );
+ e1.setField( new BooleanField( "single", true ) );
+
+ final MvccEntity mvccEntity1 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.COMPLETE, e1 );
- Entity e2 = manager.load( e1.getId() ).toBlocking().lastOrDefault( null );
+ Entity e2 = new Entity( entityId );
e2.setField( new StringField( "name", "name2" ) );
e2.setField( new IntegerField( "count", 2 ) );
e2.setField( new StringField( "nickname", "buddy" ) );
- Observable<Entity> o2 = manager.write( e2 );
- e2 = o2.toBlocking().lastOrDefault( null );
+ e2.setField( new BooleanField( "cool", false ) );
+
+ final MvccEntity mvccEntity2 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.DELETED, e2 );
- Entity e3 = manager.load( e1.getId() ).toBlocking().lastOrDefault( null );
+
+ Entity e3 = new Entity( entityId );
e3.setField( new StringField( "name", "name3" ) );
e3.setField( new IntegerField( "count", 2 ) );
- e3.getFields().remove(new StringField( "nickname", "buddy"));
- Observable<Entity> o3 = manager.write( e3 );
- e3 = o3.toBlocking().lastOrDefault( null );
-
- {
- // test minVersion of e3
- //
- // based on that data we expect something like this:
- //
- // Type = PROPERTY_WRITE, Property = count, Value = 2, Versions = [560c7e10-a925-11e3-bf9d-10ddb1de66c4]
- // Type = PROPERTY_WRITE, Property = name, Value = name3, Versions = [560c7e10-a925-11e3-bf9d-10ddb1de66c4]
- //
- // Type = PROPERTY_DELETE, Property = nickname, Value = buddy, Versions = [560b6c9e-a925-11e3-bf9d-10ddb1de66c4]
- // Type = PROPERTY_DELETE, Property = name, Value = name2, Versions = [560b6c9e-a925-11e3-bf9d-10ddb1de66c4]
- // Type = PROPERTY_DELETE, Property = count, Value = 1, Versions = [55faa3bc-a925-11e3-bf9d-10ddb1de66c4]
- // Type = PROPERTY_DELETE, Property = name, Value = name1, Versions = [55faa3bc-a925-11e3-bf9d-10ddb1de66c4]
-
- Iterator<MvccEntity> versions = mvccEntitySerializationStrategy
- .load( context, e1.getId(), e3.getVersion(), 10);
-
- ChangeLogGeneratorImpl instance = new ChangeLogGeneratorImpl();
- List<ChangeLogEntry> result = instance.getChangeLog( versions, e3.getVersion() ); // minVersion = e3
-
- for (ChangeLogEntry cle : result) {
- LOG.info( cle.toString() );
- Assert.assertFalse( cle.getVersions().isEmpty() );
- }
-
- Assert.assertEquals( 6, result.size() );
- Assert.assertTrue( isAscendingOrder( result ) );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_WRITE, result.get( 0 ).getChangeType() );
- Assert.assertEquals( "count", result.get( 0 ).getField().getName() );
- Assert.assertEquals( "2", result.get( 0 ).getField().getValue().toString() );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_WRITE, result.get( 1 ).getChangeType() );
- Assert.assertEquals( "name", result.get( 1 ).getField().getName() );
- Assert.assertEquals( "name3", result.get( 1 ).getField().getValue() );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_DELETE, result.get( 2 ).getChangeType() );
- Assert.assertEquals( "nickname", result.get( 2 ).getField().getName() );
- Assert.assertEquals( "buddy", result.get( 2 ).getField().getValue() );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_DELETE, result.get( 3 ).getChangeType() );
- Assert.assertEquals( "name", result.get( 3 ).getField().getName() );
- Assert.assertEquals( "name2", result.get( 3 ).getField().getValue() );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_DELETE, result.get( 4 ).getChangeType() );
- Assert.assertEquals( "count", result.get( 4 ).getField().getName() );
- Assert.assertEquals( "1", result.get( 4 ).getField().getValue().toString() );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_DELETE, result.get( 5 ).getChangeType() );
- Assert.assertEquals( "name", result.get( 5 ).getField().getName() );
- Assert.assertEquals( "name1", result.get( 5 ).getField().getValue() );
- }
-
- {
-
- // test minVersion of e2
- //
- // based on that data we expect something like this:
- //
- // Type = PROPERTY_WRITE, Property = name, Value = name3, Versions = [c771f63f-a927-11e3-8bfc-10ddb1de66c4]
- // Type = PROPERTY_WRITE, Property = count, Value = 2, Versions = [c770e4cd-a927-11e3-8bfc-10ddb1de66c4, c771f63f-a927-11e3-8bfc-10ddb1de66c4]
- // Type = PROPERTY_WRITE, Property = nickname, Value = buddy, Versions = [c770e4cd-a927-11e3-8bfc-10ddb1de66c4]
- // Type = PROPERTY_WRITE, Property = name, Value = name2, Versions = [c770e4cd-a927-11e3-8bfc-10ddb1de66c4]
-
- // Type = PROPERTY_DELETE, Property = count, Value = 1, Versions = [c75f589b-a927-11e3-8bfc-10ddb1de66c4]
- // Type = PROPERTY_DELETE, Property = name, Value = name1, Versions = [c75f589b-a927-11e3-8bfc-10ddb1de66c4]
-
- Iterator<MvccEntity> versions = mvccEntitySerializationStrategy
- .load( context, e1.getId(), e3.getVersion(), 10);
-
- ChangeLogGeneratorImpl instance = new ChangeLogGeneratorImpl();
- List<ChangeLogEntry> result = instance.getChangeLog( versions, e2.getVersion() ); // minVersion = e2
-
- for (ChangeLogEntry cle : result) {
- LOG.info( cle.toString() );
- Assert.assertFalse( cle.getVersions().isEmpty() );
- }
- Assert.assertEquals(6, result.size() );
- Assert.assertTrue( isAscendingOrder( result ) );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_WRITE, result.get( 2 ).getChangeType() );
- Assert.assertEquals( "nickname", result.get( 2 ).getField().getName() );
- Assert.assertEquals( "buddy", result.get( 2 ).getField().getValue() );
-
- Assert.assertEquals( ChangeLogEntry.ChangeType.PROPERTY_DELETE, result.get( 4 ).getChangeType() );
- Assert.assertEquals( "count", result.get( 4 ).getField().getName() );
- Assert.assertEquals( "1", result.get( 4 ).getField().getValue().toString() );
- }
- }
+ //appears in e1, since it's been added again, we want to make sure it doesn't appear in the delete list
+ e3.setField( new BooleanField( "single", true ) );
+
+ final MvccEntity mvccEntity3 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, e3 );
+
+
+ ChangeLogGeneratorImpl instance = new ChangeLogGeneratorImpl();
+ ChangeLog result =
+ instance.getChangeLog( Arrays.asList( mvccEntity1, mvccEntity2, mvccEntity3 ) ); // minVersion = e3
+
+
+ assertEquals( "All changes present", 0, result.getSize() );
+
+
+ Collection<Field> changes = result.getWrites();
- public static boolean isAscendingOrder( Collection<ChangeLogEntry> col ) {
- Comparable previous = null;
- for ( Comparable item : col ) {
- if ( previous == null ) {
- previous = item;
- continue;
- }
- int comparedToPrevious = item.compareTo( previous );
- if ( comparedToPrevious < 0 ) {
- return false;
- }
- }
- return true;
+ assertEquals( 0, changes.size() );
+
+ Set<String> deletes = result.getDeletes();
+
+ assertEquals( 0, deletes.size() );
}
- @SuppressWarnings( "UnusedDeclaration" )
- public static class TestModule extends JukitoModule {
+ /**
+ * Test rolling up 3 versions, properties are added then deleted
+ */
+ @Test
+ public void testDeletedLast() throws ConnectionException {
+
+ final Id entityId = new SimpleId( "test" );
+
+ Entity e1 = new Entity( entityId );
+ e1.setField( new StringField( "name", "name1" ) );
+ e1.setField( new IntegerField( "count", 1 ) );
+ e1.setField( new BooleanField( "single", true ) );
+
+ final MvccEntity mvccEntity1 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.COMPLETE, e1 );
+
+ Entity e2 = new Entity( entityId );
+ e2.setField( new StringField( "name", "name2" ) );
+ e2.setField( new IntegerField( "count", 2 ) );
+ e2.setField( new StringField( "nickname", "buddy" ) );
+ e2.setField( new BooleanField( "cool", false ) );
+
+ final MvccEntity mvccEntity2 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.PARTIAL, e2 );
+
+
+ final MvccEntity mvccEntity3 =
+ new MvccEntityImpl( entityId, UUIDGenerator.newTimeUUID(), MvccEntity.Status.DELETED,
+ Optional.<Entity>absent() );
+
+
+ ChangeLogGeneratorImpl instance = new ChangeLogGeneratorImpl();
+ ChangeLog result =
+ instance.getChangeLog( Arrays.asList( mvccEntity1, mvccEntity2, mvccEntity3 ) ); // minVersion = e3
+
+
+ assertEquals( "All changes not present", 0, result.getSize() );
+
+
+ Collection<Field> changes = result.getWrites();
+
+ assertEquals( 0, changes.size() );
+
+ Set<String> deletes = result.getDeletes();
- @Override
- protected void configureTest() {
- install( new CollectionModule() );
- }
+ assertEquals( 0, deletes.size() );
}
}