usergrid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From snoopd...@apache.org
Subject [28/34] update to master
Date Thu, 30 Jan 2014 15:21:04 GMT
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9fec2baa/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/RelationManagerImpl.java
----------------------------------------------------------------------
diff --cc stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/RelationManagerImpl.java
index b110178,0000000..8c19989
mode 100644,000000..100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/RelationManagerImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/cassandra/RelationManagerImpl.java
@@@ -1,2337 -1,0 +1,2327 @@@
 +/*******************************************************************************
 + * Copyright 2012 Apigee Corporation
 + *
 + * Licensed 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.cassandra;
 +
 +
 +import java.nio.ByteBuffer;
 +import java.util.AbstractMap;
 +import java.util.ArrayList;
 +import java.util.Collections;
 +import java.util.Iterator;
 +import java.util.LinkedHashMap;
 +import java.util.List;
 +import java.util.Map;
 +import java.util.Map.Entry;
 +import java.util.Set;
 +import java.util.TreeMap;
 +import java.util.TreeSet;
 +import java.util.UUID;
 +
 +import org.apache.usergrid.persistence.CollectionRef;
 +import org.apache.usergrid.persistence.ConnectedEntityRef;
 +import org.apache.usergrid.persistence.ConnectionRef;
 +import org.apache.usergrid.persistence.Entity;
 +import org.apache.usergrid.persistence.EntityRef;
 +import org.apache.usergrid.persistence.IndexBucketLocator;
 +import org.apache.usergrid.persistence.PagingResultsIterator;
 +import org.apache.usergrid.persistence.Query;
 +import org.apache.usergrid.persistence.RelationManager;
 +import org.apache.usergrid.persistence.Results;
 +import org.apache.usergrid.persistence.RoleRef;
 +import org.apache.usergrid.persistence.Schema;
 +import org.apache.usergrid.persistence.SimpleCollectionRef;
 +import org.apache.usergrid.persistence.SimpleEntityRef;
 +import org.apache.usergrid.persistence.SimpleRoleRef;
 +import org.apache.usergrid.persistence.IndexBucketLocator.IndexType;
 +import org.apache.usergrid.persistence.Results.Level;
 +import org.apache.usergrid.persistence.cassandra.IndexUpdate.IndexEntry;
 +import org.apache.usergrid.persistence.cassandra.index.ConnectedIndexScanner;
 +import org.apache.usergrid.persistence.cassandra.index.IndexBucketScanner;
 +import org.apache.usergrid.persistence.cassandra.index.IndexScanner;
 +import org.apache.usergrid.persistence.cassandra.index.NoOpIndexScanner;
 +import org.apache.usergrid.persistence.entities.Group;
 +import org.apache.usergrid.persistence.geo.CollectionGeoSearch;
 +import org.apache.usergrid.persistence.geo.ConnectionGeoSearch;
 +import org.apache.usergrid.persistence.geo.EntityLocationRef;
 +import org.apache.usergrid.persistence.geo.model.Point;
 +import org.apache.usergrid.persistence.query.ir.AllNode;
 +import org.apache.usergrid.persistence.query.ir.NameIdentifierNode;
 +import org.apache.usergrid.persistence.query.ir.QueryNode;
 +import org.apache.usergrid.persistence.query.ir.QuerySlice;
 +import org.apache.usergrid.persistence.query.ir.SearchVisitor;
 +import org.apache.usergrid.persistence.query.ir.WithinNode;
 +import org.apache.usergrid.persistence.query.ir.result.CollectionResultsLoaderFactory;
 +import org.apache.usergrid.persistence.query.ir.result.ConnectionIndexSliceParser;
 +import org.apache.usergrid.persistence.query.ir.result.ConnectionResultsLoaderFactory;
 +import org.apache.usergrid.persistence.query.ir.result.ConnectionTypesIterator;
 +import org.apache.usergrid.persistence.query.ir.result.EmptyIterator;
 +import org.apache.usergrid.persistence.query.ir.result.GeoIterator;
 +import org.apache.usergrid.persistence.query.ir.result.SliceIterator;
 +import org.apache.usergrid.persistence.query.ir.result.StaticIdIterator;
 +import org.apache.usergrid.persistence.query.ir.result.UUIDIndexSliceParser;
 +import org.apache.usergrid.persistence.schema.CollectionInfo;
 +import org.apache.usergrid.utils.IndexUtils;
 +import org.apache.usergrid.utils.MapUtils;
 +import org.slf4j.Logger;
 +import org.slf4j.LoggerFactory;
 +import org.springframework.util.Assert;
 +
 +import com.yammer.metrics.annotation.Metered;
 +
 +import me.prettyprint.cassandra.serializers.ByteBufferSerializer;
 +import me.prettyprint.cassandra.serializers.StringSerializer;
 +import me.prettyprint.cassandra.serializers.UUIDSerializer;
 +import me.prettyprint.hector.api.Keyspace;
 +import me.prettyprint.hector.api.beans.DynamicComposite;
 +import me.prettyprint.hector.api.beans.HColumn;
 +import me.prettyprint.hector.api.mutation.Mutator;
 +import static java.lang.String.CASE_INSENSITIVE_ORDER;
 +import static java.util.Arrays.asList;
 +import static me.prettyprint.hector.api.factory.HFactory.createMutator;
 +import static org.apache.usergrid.persistence.Schema.COLLECTION_ROLES;
 +import static org.apache.usergrid.persistence.Schema.DICTIONARY_COLLECTIONS;
 +import static org.apache.usergrid.persistence.Schema.DICTIONARY_CONNECTED_ENTITIES;
 +import static org.apache.usergrid.persistence.Schema.DICTIONARY_CONNECTED_TYPES;
 +import static org.apache.usergrid.persistence.Schema.DICTIONARY_CONNECTING_ENTITIES;
 +import static org.apache.usergrid.persistence.Schema.DICTIONARY_CONNECTING_TYPES;
 +import static org.apache.usergrid.persistence.Schema.INDEX_CONNECTIONS;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_COLLECTION_NAME;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_INACTIVITY;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_ITEM;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_ITEM_TYPE;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_NAME;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_TITLE;
 +import static org.apache.usergrid.persistence.Schema.PROPERTY_TYPE;
 +import static org.apache.usergrid.persistence.Schema.TYPE_APPLICATION;
 +import static org.apache.usergrid.persistence.Schema.TYPE_ENTITY;
 +import static org.apache.usergrid.persistence.Schema.TYPE_MEMBER;
 +import static org.apache.usergrid.persistence.Schema.TYPE_ROLE;
 +import static org.apache.usergrid.persistence.Schema.defaultCollectionName;
 +import static org.apache.usergrid.persistence.Schema.getDefaultSchema;
 +import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_COMPOSITE_DICTIONARIES;
 +import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_DICTIONARIES;
 +import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_ID_SETS;
 +import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_INDEX;
 +import static org.apache.usergrid.persistence.cassandra.ApplicationCF.ENTITY_INDEX_ENTRIES;
 +import static org.apache.usergrid.persistence.cassandra.CassandraPersistenceUtils.addDeleteToMutator;
 +import static org.apache.usergrid.persistence.cassandra.CassandraPersistenceUtils.addInsertToMutator;
 +import static org.apache.usergrid.persistence.cassandra.CassandraPersistenceUtils.batchExecute;
 +import static org.apache.usergrid.persistence.cassandra.CassandraPersistenceUtils.key;
 +import static org.apache.usergrid.persistence.cassandra.CassandraService.INDEX_ENTRY_LIST_COUNT;
 +import static org.apache.usergrid.persistence.cassandra.ConnectionRefImpl.CONNECTION_ENTITY_CONNECTION_TYPE;
 +import static org.apache.usergrid.persistence.cassandra.GeoIndexManager.batchDeleteLocationInConnectionsIndex;
 +import static org.apache.usergrid.persistence.cassandra.GeoIndexManager.batchRemoveLocationFromCollectionIndex;
 +import static org.apache.usergrid.persistence.cassandra.GeoIndexManager.batchStoreLocationInCollectionIndex;
 +import static org.apache.usergrid.persistence.cassandra.GeoIndexManager.batchStoreLocationInConnectionsIndex;
 +import static org.apache.usergrid.persistence.cassandra.IndexUpdate.indexValueCode;
 +import static org.apache.usergrid.persistence.cassandra.IndexUpdate.toIndexableValue;
 +import static org.apache.usergrid.persistence.cassandra.IndexUpdate.validIndexableValue;
 +import static org.apache.usergrid.utils.ClassUtils.cast;
 +import static org.apache.usergrid.utils.CompositeUtils.setGreaterThanEqualityFlag;
 +import static org.apache.usergrid.utils.ConversionUtils.string;
 +import static org.apache.usergrid.utils.InflectionUtils.singularize;
 +import static org.apache.usergrid.utils.MapUtils.addMapSet;
 +import static org.apache.usergrid.utils.UUIDUtils.getTimestampInMicros;
 +import static org.apache.usergrid.utils.UUIDUtils.newTimeUUID;
 +
 +
 +public class RelationManagerImpl implements RelationManager {
 +
 +    private static final Logger logger = LoggerFactory.getLogger( RelationManagerImpl.class );
 +
 +    private EntityManagerImpl em;
 +    private CassandraService cass;
 +    private UUID applicationId;
 +    private EntityRef headEntity;
 +    private IndexBucketLocator indexBucketLocator;
 +
 +    public static final StringSerializer se = new StringSerializer();
 +    public static final ByteBufferSerializer be = new ByteBufferSerializer();
 +    public static final UUIDSerializer ue = new UUIDSerializer();
 +
 +
 +    public RelationManagerImpl() {
 +    }
 +
 +
 +    public RelationManagerImpl init( EntityManagerImpl em, CassandraService cass, UUID applicationId,
 +                                     EntityRef headEntity, IndexBucketLocator indexBucketLocator ) {
 +
 +        Assert.notNull( em, "Entity manager cannot be null" );
 +        Assert.notNull( cass, "Cassandra service cannot be null" );
 +        Assert.notNull( applicationId, "Application Id cannot be null" );
 +        Assert.notNull( headEntity, "Head entity cannot be null" );
 +        Assert.notNull( headEntity.getUuid(), "Head entity uuid cannot be null" );
 +        Assert.notNull( indexBucketLocator, "Index bucket locator cannot be null" );
 +
 +        this.em = em;
 +        this.applicationId = applicationId;
 +        this.cass = cass;
 +        this.headEntity = headEntity;
 +        this.indexBucketLocator = indexBucketLocator;
 +
 +        return this;
 +    }
 +
 +
 +    private RelationManagerImpl getRelationManager( EntityRef headEntity ) {
 +        RelationManagerImpl rmi = new RelationManagerImpl();
 +        rmi.init( em, cass, applicationId, headEntity, indexBucketLocator );
 +        return rmi;
 +    }
 +
 +
 +    /** side effect: converts headEntity into an Entity if it is an EntityRef! */
 +    private Entity getHeadEntity() throws Exception {
 +        Entity entity = null;
 +        if ( headEntity instanceof Entity ) {
 +            entity = ( Entity ) headEntity;
 +        }
 +        else {
 +            entity = em.get( headEntity );
 +            headEntity = entity;
 +        }
 +        return entity;
 +    }
 +
 +
 +    /**
 +     * Batch update collection index.
 +     *
 +     * @param indexUpdate The update to apply
 +     * @param owner The entity that is the owner context of this entity update.  Can either be an application, or
 +     * another entity
 +     * @param collectionName the collection name
 +     *
 +     * @return The indexUpdate with batch mutations
 +     *
 +     * @throws Exception the exception
 +     */
 +    @Metered(group = "core", name = "RelationManager_batchUpdateCollectionIndex")
 +    public IndexUpdate batchUpdateCollectionIndex( IndexUpdate indexUpdate, EntityRef owner, String collectionName )
 +            throws Exception {
 +
 +        logger.debug( "batchUpdateCollectionIndex" );
 +
 +        Entity indexedEntity = indexUpdate.getEntity();
 +
 +        String bucketId = indexBucketLocator
 +                .getBucket( applicationId, IndexType.COLLECTION, indexedEntity.getUuid(), indexedEntity.getType(),
 +                        indexUpdate.getEntryName() );
 +
 +        // the root name without the bucket
 +        // entity_id,collection_name,prop_name,
 +        Object index_name = null;
 +        // entity_id,collection_name,prop_name, bucketId
 +        Object index_key = null;
 +
 +        // entity_id,collection_name,collected_entity_id,prop_name
 +
 +        for ( IndexEntry entry : indexUpdate.getPrevEntries() ) {
 +
 +            if ( entry.getValue() != null ) {
 +
 +                index_name = key( owner.getUuid(), collectionName, entry.getPath() );
 +
 +                index_key = key( index_name, bucketId );
 +
 +                addDeleteToMutator( indexUpdate.getBatch(), ENTITY_INDEX, index_key, entry.getIndexComposite(),
 +                        indexUpdate.getTimestamp() );
 +
 +                if ( "location.coordinates".equals( entry.getPath() ) ) {
 +                    EntityLocationRef loc = new EntityLocationRef( indexUpdate.getEntity(), entry.getTimestampUuid(),
 +                            entry.getValue().toString() );
 +                    batchRemoveLocationFromCollectionIndex( indexUpdate.getBatch(), indexBucketLocator, applicationId,
 +                            index_name, loc );
 +                }
 +            }
 +            else {
 +                logger.error( "Unexpected condition - deserialized property value is null" );
 +            }
 +        }
 +
 +        if ( ( indexUpdate.getNewEntries().size() > 0 ) && ( !indexUpdate.isMultiValue() || ( indexUpdate.isMultiValue()
 +                && !indexUpdate.isRemoveListEntry() ) ) ) {
 +
 +            for ( IndexEntry indexEntry : indexUpdate.getNewEntries() ) {
 +
 +                // byte valueCode = indexEntry.getValueCode();
 +
 +                index_name = key( owner.getUuid(), collectionName, indexEntry.getPath() );
 +
 +                index_key = key( index_name, bucketId );
 +
 +                // int i = 0;
 +
 +                addInsertToMutator( indexUpdate.getBatch(), ENTITY_INDEX, index_key, indexEntry.getIndexComposite(),
 +                        null, indexUpdate.getTimestamp() );
 +
 +                if ( "location.coordinates".equals( indexEntry.getPath() ) ) {
 +                    EntityLocationRef loc =
 +                            new EntityLocationRef( indexUpdate.getEntity(), indexEntry.getTimestampUuid(),
 +                                    indexEntry.getValue().toString() );
 +                    batchStoreLocationInCollectionIndex( indexUpdate.getBatch(), indexBucketLocator, applicationId,
 +                            index_name, indexedEntity.getUuid(), loc );
 +                }
 +
 +                // i++;
 +            }
 +        }
 +
 +        for ( String index : indexUpdate.getIndexesSet() ) {
 +            addInsertToMutator( indexUpdate.getBatch(), ENTITY_DICTIONARIES,
 +                    key( owner.getUuid(), collectionName, Schema.DICTIONARY_INDEXES ), index, null,
 +                    indexUpdate.getTimestamp() );
 +        }
 +
 +        return indexUpdate;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_getCollectionIndexes")
 +    public Set<String> getCollectionIndexes( String collectionName ) throws Exception {
 +
 +        // TODO TN, read all buckets here
 +        List<HColumn<String, String>> results =
 +                cass.getAllColumns( cass.getApplicationKeyspace( applicationId ), ENTITY_DICTIONARIES,
 +                        key( headEntity.getUuid(), collectionName, Schema.DICTIONARY_INDEXES ), se, se );
 +        Set<String> indexes = new TreeSet<String>();
 +        if ( results != null ) {
 +            for ( HColumn<String, String> column : results ) {
 +                String propertyName = column.getName();
 +                if ( !propertyName.endsWith( ".keywords" ) ) {
 +                    indexes.add( column.getName() );
 +                }
 +            }
 +        }
 +        return indexes;
 +    }
 +
 +
 +    public Map<EntityRef, Set<String>> getContainingCollections() throws Exception {
 +        Map<EntityRef, Set<String>> results = new LinkedHashMap<EntityRef, Set<String>>();
 +
 +        Keyspace ko = cass.getApplicationKeyspace( applicationId );
 +
 +        // TODO TN get all buckets here
 +
 +        List<HColumn<DynamicComposite, ByteBuffer>> containers = cass.getAllColumns( ko, ENTITY_COMPOSITE_DICTIONARIES,
 +                key( headEntity.getUuid(), Schema.DICTIONARY_CONTAINER_ENTITIES ), EntityManagerFactoryImpl.dce, be );
 +        if ( containers != null ) {
 +            for ( HColumn<DynamicComposite, ByteBuffer> container : containers ) {
 +                DynamicComposite composite = container.getName();
 +                if ( composite != null ) {
 +                    String ownerType = ( String ) composite.get( 0 );
 +                    String collectionName = ( String ) composite.get( 1 );
 +                    UUID ownerId = ( UUID ) composite.get( 2 );
 +                    addMapSet( results, new SimpleEntityRef( ownerType, ownerId ), collectionName );
 +                    if ( logger.isDebugEnabled() ) {
 +                        logger.debug( " {} ( {} ) is in collection {} ( {} ).", new Object[] {
 +                                headEntity.getType(), headEntity.getUuid(), ownerType, collectionName, ownerId
 +                        } );
 +                    }
 +                }
 +            }
 +        }
 +        EntityRef applicationRef = new SimpleEntityRef( TYPE_APPLICATION, applicationId );
 +        if ( !results.containsKey( applicationRef ) ) {
 +            addMapSet( results, applicationRef, defaultCollectionName( headEntity.getType() ) );
 +        }
 +        return results;
 +    }
 +
 +
 +    @SuppressWarnings("unchecked")
 +    public void batchCreateCollectionMembership( Mutator<ByteBuffer> batch, EntityRef ownerRef, String collectionName,
 +                                                 EntityRef itemRef, EntityRef membershipRef, UUID timestampUuid )
 +            throws Exception {
 +
 +        long timestamp = getTimestampInMicros( timestampUuid );
 +
 +        if ( membershipRef == null ) {
 +            membershipRef = new SimpleCollectionRef( ownerRef, collectionName, itemRef );
 +        }
 +
 +        Map<String, Object> properties = new TreeMap<String, Object>( CASE_INSENSITIVE_ORDER );
 +        properties.put( PROPERTY_TYPE, membershipRef.getType() );
 +        properties.put( PROPERTY_COLLECTION_NAME, collectionName );
 +        properties.put( PROPERTY_ITEM, itemRef.getUuid() );
 +        properties.put( PROPERTY_ITEM_TYPE, itemRef.getType() );
 +
 +        em.batchCreate( batch, membershipRef.getType(), null, properties, membershipRef.getUuid(), timestampUuid );
 +
 +        addInsertToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                key( membershipRef.getUuid(), Schema.DICTIONARY_CONTAINER_ENTITIES ),
 +                asList( ownerRef.getType(), collectionName, ownerRef.getUuid() ), membershipRef.getUuid(), timestamp );
 +    }
 +
 +
 +    /**
 +     * Batch add to collection.
 +     *
 +     * @param batch the batch
 +     * @param collectionName the collection name
 +     * @param entity The entity to add to the batch
 +     * @param timestampUuid The timestamp of this update in a time uuid
 +     *
 +     * @return batch
 +     *
 +     * @throws Exception the exception
 +     */
 +    public Mutator<ByteBuffer> batchAddToCollection( Mutator<ByteBuffer> batch, String collectionName, Entity entity,
 +                                                     UUID timestampUuid ) throws Exception {
 +        List<UUID> ids = new ArrayList<UUID>( 1 );
 +        ids.add( headEntity.getUuid() );
 +        return batchAddToCollections( batch, headEntity.getType(), ids, collectionName, entity, timestampUuid );
 +    }
 +
 +
 +    @SuppressWarnings("unchecked")
 +    @Metered(group = "core", name = "RelationManager_batchAddToCollections")
 +    public Mutator<ByteBuffer> batchAddToCollections( Mutator<ByteBuffer> batch, String ownerType, List<UUID> ownerIds,
 +                                                      String collectionName, Entity entity, UUID timestampUuid )
 +            throws Exception {
 +
 +        long timestamp = getTimestampInMicros( timestampUuid );
 +
 +        if ( Schema.isAssociatedEntityType( entity.getType() ) ) {
 +            logger.error( "Cant add an extended type to any collection", new Throwable() );
 +            return batch;
 +        }
 +
 +        Map<UUID, CollectionRef> membershipRefs = new LinkedHashMap<UUID, CollectionRef>();
 +
 +        for ( UUID ownerId : ownerIds ) {
 +
 +            CollectionRef membershipRef =
 +                    new SimpleCollectionRef( new SimpleEntityRef( ownerType, ownerId ), collectionName, entity );
 +
 +            membershipRefs.put( ownerId, membershipRef );
 +
 +            // get the bucket this entityId needs to be inserted into
 +            String bucketId = indexBucketLocator
 +                    .getBucket( applicationId, IndexType.COLLECTION, entity.getUuid(), collectionName );
 +
 +            Object collections_key = key( ownerId, Schema.DICTIONARY_COLLECTIONS, collectionName, bucketId );
 +
 +            // Insert in main collection
 +
 +            addInsertToMutator( batch, ENTITY_ID_SETS, collections_key, entity.getUuid(), membershipRef.getUuid(),
 +                    timestamp );
 +
 +            addInsertToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                    key( entity.getUuid(), Schema.DICTIONARY_CONTAINER_ENTITIES ),
 +                    asList( ownerType, collectionName, ownerId ), membershipRef.getUuid(), timestamp );
 +        }
 +
 +
 +        Schema schema = getDefaultSchema();
 +
 +        // Add property indexes
 +        for ( String propertyName : entity.getProperties().keySet() ) {
 +            boolean indexed_property = schema.isPropertyIndexed( entity.getType(), propertyName );
 +            if ( indexed_property ) {
 +                boolean collection_indexes_property =
 +                        schema.isPropertyIndexedInCollection( ownerType, collectionName, propertyName );
 +                boolean item_schema_has_property = schema.hasProperty( entity.getType(), propertyName );
 +                boolean fulltext_indexed = schema.isPropertyFulltextIndexed( entity.getType(), propertyName );
 +                if ( collection_indexes_property || !item_schema_has_property ) {
 +                    Object propertyValue = entity.getProperty( propertyName );
 +                    IndexUpdate indexUpdate =
 +                            batchStartIndexUpdate( batch, entity, propertyName, propertyValue, timestampUuid,
 +                                    item_schema_has_property, false, false, fulltext_indexed, true );
 +                    for ( UUID ownerId : ownerIds ) {
 +                        EntityRef owner = new SimpleEntityRef( ownerType, ownerId );
 +                        batchUpdateCollectionIndex( indexUpdate, owner, collectionName );
 +                    }
 +                }
 +            }
 +        }
 +
 +        // Add set property indexes
 +
 +        Set<String> dictionaryNames = em.getDictionaryNames( entity );
 +
 +        for ( String dictionaryName : dictionaryNames ) {
 +            boolean has_dictionary = schema.hasDictionary( entity.getType(), dictionaryName );
 +            boolean dictionary_indexed =
 +                    schema.isDictionaryIndexedInCollection( ownerType, collectionName, dictionaryName );
 +
 +            if ( dictionary_indexed || !has_dictionary ) {
 +                Set<Object> elementValues = em.getDictionaryAsSet( entity, dictionaryName );
 +                for ( Object elementValue : elementValues ) {
 +                    IndexUpdate indexUpdate =
 +                            batchStartIndexUpdate( batch, entity, dictionaryName, elementValue, timestampUuid,
 +                                    has_dictionary, true, false, false, true );
 +                    for ( UUID ownerId : ownerIds ) {
 +                        EntityRef owner = new SimpleEntityRef( ownerType, ownerId );
 +                        batchUpdateCollectionIndex( indexUpdate, owner, collectionName );
 +                    }
 +                }
 +            }
 +        }
 +
 +        for ( UUID ownerId : ownerIds ) {
 +            EntityRef owner = new SimpleEntityRef( ownerType, ownerId );
 +            batchCreateCollectionMembership( batch, owner, collectionName, entity, membershipRefs.get( ownerId ),
 +                    timestampUuid );
 +        }
 +
 +        return batch;
 +    }
 +
 +
 +    /**
 +     * Batch remove from collection.
 +     * <p/>
 +     * * Batch add to collection.
 +     *
 +     * @param batch the batch
 +     * @param collectionName the collection name
 +     * @param entity The entity to add to the batch
 +     * @param timestampUuid The timestamp of this update in a time uuid
 +     *
 +     * @return The mutation with the delete operations added
 +     *
 +     * @throws Exception the exception
 +     */
 +    public Mutator<ByteBuffer> batchRemoveFromCollection( Mutator<ByteBuffer> batch, String collectionName,
 +                                                          Entity entity, UUID timestampUuid ) throws Exception {
 +        return this.batchRemoveFromCollection( batch, collectionName, entity, false, timestampUuid );
 +    }
 +
 +
 +    @SuppressWarnings("unchecked")
 +    @Metered(group = "core", name = "RelationManager_batchRemoveFromCollection")
 +    public Mutator<ByteBuffer> batchRemoveFromCollection( Mutator<ByteBuffer> batch, String collectionName,
 +                                                          Entity entity, boolean force, UUID timestampUuid )
 +            throws Exception {
 +
 +        long timestamp = getTimestampInMicros( timestampUuid );
 +
 +        if ( !force && headEntity.getUuid().equals( applicationId ) ) {
 +            // Can't remove entities from root collections
 +            return batch;
 +        }
 +
 +        Object collections_key = key( headEntity.getUuid(), Schema.DICTIONARY_COLLECTIONS, collectionName,
 +                indexBucketLocator.getBucket( applicationId, IndexType.COLLECTION, entity.getUuid(), collectionName ) );
 +
 +        // Remove property indexes
 +
 +        Schema schema = getDefaultSchema();
 +        for ( String propertyName : entity.getProperties().keySet() ) {
 +            boolean collection_indexes_property =
 +                    schema.isPropertyIndexedInCollection( headEntity.getType(), collectionName, propertyName );
 +            boolean item_schema_has_property = schema.hasProperty( entity.getType(), propertyName );
 +            boolean fulltext_indexed = schema.isPropertyFulltextIndexed( entity.getType(), propertyName );
 +            if ( collection_indexes_property || !item_schema_has_property ) {
 +                IndexUpdate indexUpdate = batchStartIndexUpdate( batch, entity, propertyName, null, timestampUuid,
 +                        item_schema_has_property, false, false, fulltext_indexed );
 +                batchUpdateCollectionIndex( indexUpdate, headEntity, collectionName );
 +            }
 +        }
 +
 +        // Remove set indexes
 +
 +        Set<String> dictionaryNames = em.getDictionaryNames( entity );
 +
 +        for ( String dictionaryName : dictionaryNames ) {
 +            boolean has_dictionary = schema.hasDictionary( entity.getType(), dictionaryName );
 +            boolean dictionary_indexed =
 +                    schema.isDictionaryIndexedInCollection( headEntity.getType(), collectionName, dictionaryName );
 +
 +            if ( dictionary_indexed || !has_dictionary ) {
 +                Set<Object> elementValues = em.getDictionaryAsSet( entity, dictionaryName );
 +                for ( Object elementValue : elementValues ) {
 +                    IndexUpdate indexUpdate =
 +                            batchStartIndexUpdate( batch, entity, dictionaryName, elementValue, timestampUuid,
 +                                    has_dictionary, true, true, false );
 +                    batchUpdateCollectionIndex( indexUpdate, headEntity, collectionName );
 +                }
 +            }
 +        }
 +
 +        // Delete actual property
 +
 +        addDeleteToMutator( batch, ENTITY_ID_SETS, collections_key, entity.getUuid(), timestamp );
 +
 +        addDeleteToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                key( entity.getUuid(), Schema.DICTIONARY_CONTAINER_ENTITIES ),
 +                asList( headEntity.getType(), collectionName, headEntity.getUuid() ), timestamp );
 +
 +        if ( !headEntity.getType().equalsIgnoreCase( TYPE_APPLICATION ) && !Schema
 +                .isAssociatedEntityType( entity.getType() ) ) {
 +            em.deleteEntity( new SimpleCollectionRef( headEntity, collectionName, entity ).getUuid() );
 +        }
 +
 +        return batch;
 +    }
 +
 +
 +    @Metered(group = "core", name = "RelationManager_batchDeleteConnectionIndexEntries")
 +    public Mutator<ByteBuffer> batchDeleteConnectionIndexEntries( IndexUpdate indexUpdate, IndexEntry entry,
 +                                                                  ConnectionRefImpl connection, UUID[] index_keys )
 +            throws Exception {
 +
 +        // entity_id,prop_name
 +        Object property_index_key = key( index_keys[ConnectionRefImpl.ALL], INDEX_CONNECTIONS, entry.getPath(),
 +                indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION, index_keys[ConnectionRefImpl.ALL],
 +                        entry.getPath() ) );
 +
 +        // entity_id,entity_type,prop_name
 +        Object entity_type_prop_index_key =
 +                key( index_keys[ConnectionRefImpl.BY_ENTITY_TYPE], INDEX_CONNECTIONS, entry.getPath(),
 +                        indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION,
 +                                index_keys[ConnectionRefImpl.BY_ENTITY_TYPE], entry.getPath() ) );
 +
 +        // entity_id,connection_type,prop_name
 +        Object connection_type_prop_index_key =
 +                key( index_keys[ConnectionRefImpl.BY_CONNECTION_TYPE], INDEX_CONNECTIONS, entry.getPath(),
 +                        indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION,
 +                                index_keys[ConnectionRefImpl.BY_CONNECTION_TYPE], entry.getPath() ) );
 +
 +        // entity_id,connection_type,entity_type,prop_name
 +        Object connection_type_and_entity_type_prop_index_key =
 +                key( index_keys[ConnectionRefImpl.BY_CONNECTION_AND_ENTITY_TYPE], INDEX_CONNECTIONS, entry.getPath(),
 +                        indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION,
 +                                index_keys[ConnectionRefImpl.BY_CONNECTION_AND_ENTITY_TYPE], entry.getPath() ) );
 +
 +        // composite(property_value,connected_entity_id,connection_type,entity_type,entry_timestamp)
 +        addDeleteToMutator( indexUpdate.getBatch(), ENTITY_INDEX, property_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId(), connection.getConnectionType(),
 +                        connection.getConnectedEntityType() ), indexUpdate.getTimestamp() );
 +
 +        // composite(property_value,connected_entity_id,connection_type,entry_timestamp)
 +        addDeleteToMutator( indexUpdate.getBatch(), ENTITY_INDEX, entity_type_prop_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId(), connection.getConnectionType() ),
 +                indexUpdate.getTimestamp() );
 +
 +        // composite(property_value,connected_entity_id,entity_type,entry_timestamp)
 +        addDeleteToMutator( indexUpdate.getBatch(), ENTITY_INDEX, connection_type_prop_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId(), connection.getConnectedEntityType() ),
 +                indexUpdate.getTimestamp() );
 +
 +        // composite(property_value,connected_entity_id,entry_timestamp)
 +        addDeleteToMutator( indexUpdate.getBatch(), ENTITY_INDEX, connection_type_and_entity_type_prop_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId() ), indexUpdate.getTimestamp() );
 +
 +        return indexUpdate.getBatch();
 +    }
 +
 +
 +    @Metered(group = "core", name = "RelationManager_batchAddConnectionIndexEntries")
 +    public Mutator<ByteBuffer> batchAddConnectionIndexEntries( IndexUpdate indexUpdate, IndexEntry entry,
 +                                                               ConnectionRefImpl connection, UUID[] index_keys ) {
 +
 +        // entity_id,prop_name
 +        Object property_index_key = key( index_keys[ConnectionRefImpl.ALL], INDEX_CONNECTIONS, entry.getPath(),
 +                indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION, index_keys[ConnectionRefImpl.ALL],
 +                        entry.getPath() ) );
 +
 +        // entity_id,entity_type,prop_name
 +        Object entity_type_prop_index_key =
 +                key( index_keys[ConnectionRefImpl.BY_ENTITY_TYPE], INDEX_CONNECTIONS, entry.getPath(),
 +                        indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION,
 +                                index_keys[ConnectionRefImpl.BY_ENTITY_TYPE], entry.getPath() ) );
 +
 +        // entity_id,connection_type,prop_name
 +        Object connection_type_prop_index_key =
 +                key( index_keys[ConnectionRefImpl.BY_CONNECTION_TYPE], INDEX_CONNECTIONS, entry.getPath(),
 +                        indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION,
 +                                index_keys[ConnectionRefImpl.BY_CONNECTION_TYPE], entry.getPath() ) );
 +
 +        // entity_id,connection_type,entity_type,prop_name
 +        Object connection_type_and_entity_type_prop_index_key =
 +                key( index_keys[ConnectionRefImpl.BY_CONNECTION_AND_ENTITY_TYPE], INDEX_CONNECTIONS, entry.getPath(),
 +                        indexBucketLocator.getBucket( applicationId, IndexType.CONNECTION,
 +                                index_keys[ConnectionRefImpl.BY_CONNECTION_AND_ENTITY_TYPE], entry.getPath() ) );
 +
 +        // composite(property_value,connected_entity_id,connection_type,entity_type,entry_timestamp)
 +        addInsertToMutator( indexUpdate.getBatch(), ENTITY_INDEX, property_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId(), connection.getConnectionType(),
 +                        connection.getConnectedEntityType() ), connection.getUuid(), indexUpdate.getTimestamp() );
 +
 +        // composite(property_value,connected_entity_id,connection_type,entry_timestamp)
 +        addInsertToMutator( indexUpdate.getBatch(), ENTITY_INDEX, entity_type_prop_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId(), connection.getConnectionType() ),
 +                connection.getUuid(), indexUpdate.getTimestamp() );
 +
 +        // composite(property_value,connected_entity_id,entity_type,entry_timestamp)
 +        addInsertToMutator( indexUpdate.getBatch(), ENTITY_INDEX, connection_type_prop_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId(), connection.getConnectedEntityType() ),
 +                connection.getUuid(), indexUpdate.getTimestamp() );
 +
 +        // composite(property_value,connected_entity_id,entry_timestamp)
 +        addInsertToMutator( indexUpdate.getBatch(), ENTITY_INDEX, connection_type_and_entity_type_prop_index_key,
 +                entry.getIndexComposite( connection.getConnectedEntityId() ), connection.getUuid(),
 +                indexUpdate.getTimestamp() );
 +
 +        return indexUpdate.getBatch();
 +    }
 +
 +
 +    /**
 +     * Batch update connection index.
 +     *
 +     * @param indexUpdate The update operation to perform
 +     * @param connection The connection to update
 +     *
 +     * @return The index with the batch mutation udpated
 +     *
 +     * @throws Exception the exception
 +     */
 +    @Metered(group = "core", name = "RelationManager_batchUpdateConnectionIndex")
 +    public IndexUpdate batchUpdateConnectionIndex( IndexUpdate indexUpdate, ConnectionRefImpl connection )
 +            throws Exception {
 +
 +        // UUID connection_id = connection.getUuid();
 +
 +        UUID[] index_keys = connection.getIndexIds();
 +
 +        // Delete all matching entries from entry list
 +        for ( IndexEntry entry : indexUpdate.getPrevEntries() ) {
 +
 +            if ( entry.getValue() != null ) {
 +
 +                batchDeleteConnectionIndexEntries( indexUpdate, entry, connection, index_keys );
 +
 +                if ( "location.coordinates".equals( entry.getPath() ) ) {
 +                    EntityLocationRef loc = new EntityLocationRef( indexUpdate.getEntity(), entry.getTimestampUuid(),
 +                            entry.getValue().toString() );
 +                    batchDeleteLocationInConnectionsIndex( indexUpdate.getBatch(), indexBucketLocator, applicationId,
 +                            index_keys, entry.getPath(), loc );
 +                }
 +            }
 +            else {
 +                logger.error( "Unexpected condition - deserialized property value is null" );
 +            }
 +        }
 +
 +        if ( ( indexUpdate.getNewEntries().size() > 0 ) && ( !indexUpdate.isMultiValue() || ( indexUpdate.isMultiValue()
 +                && !indexUpdate.isRemoveListEntry() ) ) ) {
 +
 +            for ( IndexEntry indexEntry : indexUpdate.getNewEntries() ) {
 +
 +                batchAddConnectionIndexEntries( indexUpdate, indexEntry, connection, index_keys );
 +
 +                if ( "location.coordinates".equals( indexEntry.getPath() ) ) {
 +                    EntityLocationRef loc =
 +                            new EntityLocationRef( indexUpdate.getEntity(), indexEntry.getTimestampUuid(),
 +                                    indexEntry.getValue().toString() );
 +                    batchStoreLocationInConnectionsIndex( indexUpdate.getBatch(), indexBucketLocator, applicationId,
 +                            index_keys, indexEntry.getPath(), loc );
 +                }
 +            }
 +
 +      /*
 +       * addInsertToMutator(batch, EntityCF.SETS, key(connection_id,
 +       * Schema.INDEXES_SET), indexEntry.getKey(), null, false, timestamp); }
 +       * 
 +       * addInsertToMutator(batch, EntityCF.SETS, key(connection_id,
 +       * Schema.INDEXES_SET), entryName, null, false, timestamp);
 +       */
 +        }
 +
 +        for ( String index : indexUpdate.getIndexesSet() ) {
 +            addInsertToMutator( indexUpdate.getBatch(), ENTITY_DICTIONARIES,
 +                    key( connection.getConnectingIndexId(), Schema.DICTIONARY_INDEXES ), index, null,
 +                    indexUpdate.getTimestamp() );
 +        }
 +
 +        return indexUpdate;
 +    }
 +
 +
 +    public Set<String> getConnectionIndexes( ConnectionRefImpl connection ) throws Exception {
 +        List<HColumn<String, String>> results =
 +                cass.getAllColumns( cass.getApplicationKeyspace( applicationId ), ENTITY_DICTIONARIES,
 +                        key( connection.getConnectingIndexId(), Schema.DICTIONARY_INDEXES ), se, se );
 +        Set<String> indexes = new TreeSet<String>();
 +        if ( results != null ) {
 +            for ( HColumn<String, String> column : results ) {
 +                String propertyName = column.getName();
 +                if ( !propertyName.endsWith( ".keywords" ) ) {
 +                    indexes.add( column.getName() );
 +                }
 +            }
 +        }
 +        return indexes;
 +    }
 +
 +
 +    /**
 +     * Batch update backward connections property indexes.
 +     *
 +     * @param indexUpdate The update to run for incoming connections
 +     *
 +     * @return The index update to run
 +     *
 +     * @throws Exception the exception
 +     */
 +    @Metered(group = "core", name = "RelationManager_batchUpdateBackwardConnectionsPropertyIndexes")
 +    public IndexUpdate batchUpdateBackwardConnectionsPropertyIndexes( IndexUpdate indexUpdate ) throws Exception {
 +
 +        logger.debug( "batchUpdateBackwordConnectionsPropertyIndexes" );
 +
 +        boolean entitySchemaHasProperty = indexUpdate.isSchemaHasProperty();
 +
 +        if ( entitySchemaHasProperty ) {
 +            if ( !getDefaultSchema()
 +                    .isPropertyIndexed( indexUpdate.getEntity().getType(), indexUpdate.getEntryName() ) ) {
 +                return indexUpdate;
 +            }
 +        }
 +
 +
 +        return doBackwardConnectionsUpdate( indexUpdate );
 +    }
 +
 +
 +    /**
 +     * Search each reverse connection type in the graph for connections.  If one is found, update the index
 +     * appropriately
 +     *
 +     * @param indexUpdate The index update to use
 +     *
 +     * @return The updated index update
 +     */
 +    private IndexUpdate doBackwardConnectionsUpdate( IndexUpdate indexUpdate ) throws Exception {
 +        final Entity targetEntity = indexUpdate.getEntity();
 +
 +        final ConnectionTypesIterator connectionTypes =
 +                new ConnectionTypesIterator( cass, applicationId, targetEntity.getUuid(), false, 100 );
 +
 +        for ( String connectionType : connectionTypes ) {
 +
 +            PagingResultsIterator itr = getReversedConnectionsIterator( targetEntity, connectionType );
 +
 +            for ( Object connection : itr ) {
 +
 +                final ConnectedEntityRef sourceEntity = ( ConnectedEntityRef ) connection;
 +
 +                //we need to create a connection ref from the source entity (found via reverse edge) to the entity
 +                // we're about to update.  This is the index that needs updated
 +                final ConnectionRefImpl connectionRef =
 +                        new ConnectionRefImpl( sourceEntity, connectionType, indexUpdate.getEntity() );
 +
 +                batchUpdateConnectionIndex( indexUpdate, connectionRef );
 +            }
 +        }
 +
 +        return indexUpdate;
 +    }
 +
 +
 +    /**
 +     * Get a paging results iterator.  Should return an iterator for all results
 +     *
 +     * @param targetEntity The target entity search connections from
 +     *
 +     * @return connectionType The name of the edges to search
 +     */
 +    private PagingResultsIterator getReversedConnectionsIterator( EntityRef targetEntity, String connectionType )
 +            throws Exception {
 +        return new PagingResultsIterator( getConnectingEntities( targetEntity, connectionType, null, Level.REFS ) );
 +    }
 +
 +
 +    /**
 +     * Batch update backward connections set indexes.
 +     *
 +     * @param indexUpdate The index to update in the dictionary
 +     *
 +     * @return The index update
 +     *
 +     * @throws Exception the exception
 +     */
 +    @Metered(group = "core", name = "RelationManager_batchUpdateBackwardConnectionsDictionaryIndexes")
 +    public IndexUpdate batchUpdateBackwardConnectionsDictionaryIndexes( IndexUpdate indexUpdate ) throws Exception {
 +
 +        logger.debug( "batchUpdateBackwardConnectionsListIndexes" );
 +
 +        boolean entityHasDictionary = getDefaultSchema()
 +                .isDictionaryIndexedInConnections( indexUpdate.getEntity().getType(), indexUpdate.getEntryName() );
 +
 +        if ( !entityHasDictionary ) {
 +            return indexUpdate;
 +        }
 +
 +
 +        return doBackwardConnectionsUpdate( indexUpdate );
 +    }
 +
 +
 +    @SuppressWarnings("unchecked")
 +    @Metered(group = "core", name = "RelationManager_batchUpdateEntityConnection")
 +    public Mutator<ByteBuffer> batchUpdateEntityConnection( Mutator<ByteBuffer> batch, boolean disconnect,
 +                                                            ConnectionRefImpl connection, UUID timestampUuid )
 +            throws Exception {
 +
 +        long timestamp = getTimestampInMicros( timestampUuid );
 +
 +        Entity connectedEntity = em.get( connection.getConnectedEntityId() );
 +
 +        if ( connectedEntity == null ) {
 +            return batch;
 +        }
 +
 +        // Create connection for requested params
 +
 +
 +        if ( disconnect ) {
 +            addDeleteToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                    key( connection.getConnectingEntityId(), DICTIONARY_CONNECTED_ENTITIES,
 +                            connection.getConnectionType() ),
 +                    asList( connection.getConnectedEntityId(), connection.getConnectedEntityType() ), timestamp );
 +
 +            addDeleteToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                    key( connection.getConnectedEntityId(), DICTIONARY_CONNECTING_ENTITIES,
 +                            connection.getConnectionType() ),
 +                    asList( connection.getConnectingEntityId(), connection.getConnectingEntityType() ), timestamp );
 +
 +            // delete the connection path if there will be no connections left
 +
 +            boolean delete = true;
 +
 +            //check out outbound edges of the given type.  If we have more than the 1 specified,
 +            // we shouldn't delete the connection types from our outbound index
 +            PagingResultsIterator itr = new PagingResultsIterator(
 +                    getConnectedEntities( connection.getConnectingEntity(), connection.getConnectionType(), null,
 +                            Level.REFS ) );
 +
 +            ConnectedEntityRef c;
 +
 +            while ( itr.hasNext() ) {
 +                c = ( ConnectedEntityRef ) itr.next();
 +
 +                if ( !connection.getConnectedEntityId().equals( c.getUuid() ) ) {
 +                    delete = false;
 +                    break;
 +                }
 +
 +
 +                //        c = (ConnectionRef) itr.next();
 +                //        if (c.getConnectedEntity().getConnectionType().equals(connection.getConnectedEntity()
 +                // .getConnectionType()) &&!c.getConnectedEntity().getUuid().equals(connection.getConnectedEntity()
 +                // .getUuid())) {
 +                //            delete = false;
 +                //            break;
 +                //        }
 +
 +            }
 +            //      for (ConnectionRefImpl c : getConnectionsWithEntity(connection.getConnectingEntityId())) {
 +            //        if (c.getConnectedEntity().getConnectionType().equals(connection.getConnectedEntity()
 +            // .getConnectionType())) {
 +            //          if (!c.getConnectedEntity().getUuid().equals(connection.getConnectedEntity().getUuid())) {
 +            //            delete = false;
 +            //            break;
 +            //          }
 +            //        }
 +            //      }
 +            if ( delete ) {
 +                addDeleteToMutator( batch, ENTITY_DICTIONARIES,
 +                        key( connection.getConnectingEntityId(), DICTIONARY_CONNECTED_TYPES ),
 +                        connection.getConnectionType(), timestamp );
 +            }
 +
 +            // delete the connection path if there will be no connections left
 +            delete = true;
 +
 +
 +            //check out inbound edges of the given type.  If we have more than the 1 specified,
 +            // we shouldn't delete the connection types from our outbound index
 +            itr = new PagingResultsIterator(
 +                    getConnectingEntities( connection.getConnectingEntity(), connection.getConnectionType(), null,
 +                            Level.REFS ) );
 +
 +            while ( itr.hasNext() ) {
 +                c = ( ConnectedEntityRef ) itr.next();
 +
 +                if ( !connection.getConnectedEntityId().equals( c.getUuid() ) ) {
 +                    delete = false;
 +                    break;
 +                }
 +                //        if (c.getConnectedEntity().getConnectionType().equals(connection.getConnectedEntity()
 +                // .getConnectionType()) && !c.getConnectingEntity().getUuid().equals(connection.getConnectingEntity
 +                // ().getUuid())) {
 +                //            delete = false;
 +                //            break;
 +                //        }
 +
 +            }
 +
 +            //      for (ConnectionRefImpl c : getConnectionsWithEntity(connection.getConnectedEntityId())) {
 +            //        if (c.getConnectedEntity().getConnectionType().equals(connection.getConnectedEntity()
 +            // .getConnectionType())) {
 +            //          if (!c.getConnectingEntity().getUuid().equals(connection.getConnectingEntity().getUuid())) {
 +            //            delete = false;
 +            //            break;
 +            //          }
 +            //        }
 +            //      }
 +            if ( delete ) {
 +                addDeleteToMutator( batch, ENTITY_DICTIONARIES,
 +                        key( connection.getConnectedEntityId(), DICTIONARY_CONNECTING_TYPES ),
 +                        connection.getConnectionType(), timestamp );
 +            }
 +        }
 +        else {
 +            addInsertToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                    key( connection.getConnectingEntityId(), DICTIONARY_CONNECTED_ENTITIES,
 +                            connection.getConnectionType() ),
 +                    asList( connection.getConnectedEntityId(), connection.getConnectedEntityType() ), timestamp,
 +                    timestamp );
 +
 +            addInsertToMutator( batch, ENTITY_COMPOSITE_DICTIONARIES,
 +                    key( connection.getConnectedEntityId(), DICTIONARY_CONNECTING_ENTITIES,
 +                            connection.getConnectionType() ),
 +                    asList( connection.getConnectingEntityId(), connection.getConnectingEntityType() ), timestamp,
 +                    timestamp );
 +
 +            // Add connection type to connections set
 +            addInsertToMutator( batch, ENTITY_DICTIONARIES,
 +                    key( connection.getConnectingEntityId(), DICTIONARY_CONNECTED_TYPES ),
 +                    connection.getConnectionType(), null, timestamp );
 +
 +            // Add connection type to connections set
 +            addInsertToMutator( batch, ENTITY_DICTIONARIES,
 +                    key( connection.getConnectedEntityId(), DICTIONARY_CONNECTING_TYPES ),
 +                    connection.getConnectionType(), null, timestamp );
 +        }
 +
 +        // Add property indexes
 +
 +        // Iterate though all the properties of the connected entity
 +
 +        Schema schema = getDefaultSchema();
 +        for ( String propertyName : connectedEntity.getProperties().keySet() ) {
 +            Object propertyValue = connectedEntity.getProperties().get( propertyName );
 +
 +            boolean indexed = schema.isPropertyIndexed( connectedEntity.getType(), propertyName );
 +
 +            boolean connection_indexes_property = schema.isPropertyIndexed( connectedEntity.getType(), propertyName );
 +            boolean item_schema_has_property = schema.hasProperty( connectedEntity.getType(), propertyName );
 +            boolean fulltext_indexed = schema.isPropertyFulltextIndexed( connectedEntity.getType(), propertyName );
 +            // For each property, if the schema says it's indexed, update its
 +            // index
 +
 +            if ( indexed && ( connection_indexes_property || !item_schema_has_property ) ) {
 +                IndexUpdate indexUpdate =
 +                        batchStartIndexUpdate( batch, connectedEntity, propertyName, disconnect ? null : propertyValue,
 +                                timestampUuid, item_schema_has_property, false, false, fulltext_indexed );
 +                batchUpdateConnectionIndex( indexUpdate, connection );
 +            }
 +        }
 +
 +        // Add indexes for the connected entity's list properties
 +
 +        // Get the names of the list properties in the connected entity
 +        Set<String> dictionaryNames = em.getDictionaryNames( connectedEntity );
 +
 +        // For each list property, get the values in the list and
 +        // update the index with those values
 +
 +        for ( String dictionaryName : dictionaryNames ) {
 +            boolean has_dictionary = schema.hasDictionary( connectedEntity.getType(), dictionaryName );
 +            boolean dictionary_indexed =
 +                    schema.isDictionaryIndexedInConnections( connectedEntity.getType(), dictionaryName );
 +
 +            if ( dictionary_indexed || !has_dictionary ) {
 +                Set<Object> elementValues = em.getDictionaryAsSet( connectedEntity, dictionaryName );
 +                for ( Object elementValue : elementValues ) {
 +                    IndexUpdate indexUpdate =
 +                            batchStartIndexUpdate( batch, connectedEntity, dictionaryName, elementValue, timestampUuid,
 +                                    has_dictionary, true, disconnect, false );
 +                    batchUpdateConnectionIndex( indexUpdate, connection );
 +                }
 +            }
 +        }
 +
 +        return batch;
 +    }
 +
 +
 +    public void updateEntityConnection( boolean disconnect, ConnectionRefImpl connection ) throws Exception {
 +
 +        UUID timestampUuid = newTimeUUID();
 +        Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
 +
 +        // Make or break the connection
 +
 +        batchUpdateEntityConnection( batch, disconnect, connection, timestampUuid );
 +
 +        // Make or break a connection from the connecting entity
 +        // to the connection itself
 +
 +        ConnectionRefImpl loopback = connection.getConnectionToConnectionEntity();
 +        if ( !disconnect ) {
 +            em.insertEntity( CONNECTION_ENTITY_CONNECTION_TYPE, loopback.getConnectedEntityId() );
 +        }
 +
 +        batchUpdateEntityConnection( batch, disconnect, loopback, timestampUuid );
 +
 +        batchExecute( batch, CassandraService.RETRY_COUNT );
 +    }
 +
 +
 +    @Metered(group = "core", name = "RelationManager_batchDisconnect")
 +    public void batchDisconnect( Mutator<ByteBuffer> batch, UUID timestampUuid ) throws Exception {
 +
 +
 +        PagingResultsIterator itr =
 +                new PagingResultsIterator( getConnectingEntities( headEntity, null, null, Level.REFS ) );
 +
 +        ConnectionRefImpl connection;
 +
 +        while ( itr.hasNext() ) {
 +            connection = ( ConnectionRefImpl ) itr.next();
 +
 +            batchUpdateEntityConnection( batch, true, connection, timestampUuid );
 +        }
 +        //
 +        //    List<ConnectionRefImpl> connections = getConnectionsWithEntity(headEntity.getUuid());
 +        //    if (connections != null) {
 +        //      for (ConnectionRefImpl connection : connections) {
 +        //        batchUpdateEntityConnection(batch, true, connection, timestampUuid);
 +        //      }
 +        //    }
 +    }
 +
 +
 +    public IndexUpdate batchStartIndexUpdate( Mutator<ByteBuffer> batch, Entity entity, String entryName,
 +                                              Object entryValue, UUID timestampUuid, boolean schemaHasProperty,
 +                                              boolean isMultiValue, boolean removeListEntry, boolean fulltextIndexed )
 +            throws Exception {
 +        return batchStartIndexUpdate( batch, entity, entryName, entryValue, timestampUuid, schemaHasProperty,
 +                isMultiValue, removeListEntry, fulltextIndexed, false );
 +    }
 +
 +
 +    @Metered(group = "core", name = "RelationManager_batchStartIndexUpdate")
 +    public IndexUpdate batchStartIndexUpdate( Mutator<ByteBuffer> batch, Entity entity, String entryName,
 +                                              Object entryValue, UUID timestampUuid, boolean schemaHasProperty,
 +                                              boolean isMultiValue, boolean removeListEntry, boolean fulltextIndexed,
 +                                              boolean skipRead ) throws Exception {
 +
 +        long timestamp = getTimestampInMicros( timestampUuid );
 +
 +        IndexUpdate indexUpdate =
 +                new IndexUpdate( batch, entity, entryName, entryValue, schemaHasProperty, isMultiValue, removeListEntry,
 +                        timestampUuid );
 +
 +        // entryName = entryName.toLowerCase();
 +
 +        // entity_id,connection_type,connected_entity_id,prop_name
 +
 +        if ( !skipRead ) {
 +
 +            List<HColumn<ByteBuffer, ByteBuffer>> entries = null;
 +
 +            if ( isMultiValue && validIndexableValue( entryValue ) ) {
 +                entries = cass.getColumns( cass.getApplicationKeyspace( applicationId ), ENTITY_INDEX_ENTRIES,
 +                        entity.getUuid(),
 +                        new DynamicComposite( entryName, indexValueCode( entryValue ), toIndexableValue( entryValue ) ),
 +                        setGreaterThanEqualityFlag( new DynamicComposite( entryName, indexValueCode( entryValue ),
 +                                toIndexableValue( entryValue ) ) ), INDEX_ENTRY_LIST_COUNT, false );
 +            }
 +            else {
 +                entries = cass.getColumns( cass.getApplicationKeyspace( applicationId ), ENTITY_INDEX_ENTRIES,
 +                        entity.getUuid(), new DynamicComposite( entryName ),
 +                        setGreaterThanEqualityFlag( new DynamicComposite( entryName ) ), INDEX_ENTRY_LIST_COUNT,
 +                        false );
 +            }
 +
 +            if ( logger.isDebugEnabled() ) {
 +                logger.debug( "Found {} previous index entries for {} of entity {}", new Object[] {
 +                        entries.size(), entryName, entity.getUuid()
 +                } );
 +            }
 +
 +            // Delete all matching entries from entry list
 +            for ( HColumn<ByteBuffer, ByteBuffer> entry : entries ) {
 +                UUID prev_timestamp = null;
 +                Object prev_value = null;
 +                String prev_obj_path = null;
 +
 +                // new format:
 +                // composite(entryName,
 +                // value_code,prev_value,prev_timestamp,prev_obj_path) = null
 +                DynamicComposite composite = DynamicComposite.fromByteBuffer( entry.getName().duplicate() );
 +                prev_value = composite.get( 2 );
 +                prev_timestamp = ( UUID ) composite.get( 3 );
 +                if ( composite.size() > 4 ) {
 +                    prev_obj_path = ( String ) composite.get( 4 );
 +                }
 +
 +                if ( prev_value != null ) {
 +
 +                    String entryPath = entryName;
 +                    if ( ( prev_obj_path != null ) && ( prev_obj_path.length() > 0 ) ) {
 +                        entryPath = entryName + "." + prev_obj_path;
 +                    }
 +
 +                    indexUpdate.addPrevEntry( entryPath, prev_value, prev_timestamp, entry.getName().duplicate() );
 +
 +                    // composite(property_value,connected_entity_id,entry_timestamp)
 +                    // addDeleteToMutator(batch, ENTITY_INDEX_ENTRIES,
 +                    // entity.getUuid(), entry.getName(), timestamp);
 +
 +                }
 +                else {
 +                    logger.error( "Unexpected condition - deserialized property value is null" );
 +                }
 +            }
 +        }
 +
 +        if ( !isMultiValue || ( isMultiValue && !removeListEntry ) ) {
 +
 +            List<Map.Entry<String, Object>> list = IndexUtils.getKeyValueList( entryName, entryValue, fulltextIndexed );
 +
 +            if ( entryName.equalsIgnoreCase( "location" ) && ( entryValue instanceof Map ) ) {
 +                @SuppressWarnings("rawtypes") double latitude =
 +                        MapUtils.getDoubleValue( ( Map ) entryValue, "latitude" );
 +                @SuppressWarnings("rawtypes") double longitude =
 +                        MapUtils.getDoubleValue( ( Map ) entryValue, "longitude" );
 +                list.add( new AbstractMap.SimpleEntry<String, Object>( "location.coordinates",
 +                        latitude + "," + longitude ) );
 +            }
 +
 +            for ( Map.Entry<String, Object> indexEntry : list ) {
 +
 +                if ( validIndexableValue( indexEntry.getValue() ) ) {
 +                    indexUpdate.addNewEntry( indexEntry.getKey(), toIndexableValue( indexEntry.getValue() ) );
 +                }
 +            }
 +
 +            if ( isMultiValue ) {
 +                addInsertToMutator( batch, ENTITY_INDEX_ENTRIES, entity.getUuid(),
 +                        asList( entryName, indexValueCode( entryValue ), toIndexableValue( entryValue ),
 +                                indexUpdate.getTimestampUuid() ), null, timestamp );
 +            }
 +            else {
 +                // int i = 0;
 +
 +                for ( Map.Entry<String, Object> indexEntry : list ) {
 +
 +                    String name = indexEntry.getKey();
 +                    if ( name.startsWith( entryName + "." ) ) {
 +                        name = name.substring( entryName.length() + 1 );
 +                    }
 +                    else if ( name.startsWith( entryName ) ) {
 +                        name = name.substring( entryName.length() );
 +                    }
 +
 +                    byte code = indexValueCode( indexEntry.getValue() );
 +                    Object val = toIndexableValue( indexEntry.getValue() );
 +                    addInsertToMutator( batch, ENTITY_INDEX_ENTRIES, entity.getUuid(),
 +                            asList( entryName, code, val, indexUpdate.getTimestampUuid(), name ), null, timestamp );
 +
 +                    indexUpdate.addIndex( indexEntry.getKey() );
 +                }
 +            }
 +
 +            indexUpdate.addIndex( entryName );
 +        }
 +
 +        return indexUpdate;
 +    }
 +
 +
 +    @Metered(group = "core", name = "RelationManager_batchUpdatePropertyIndexes")
 +    public void batchUpdatePropertyIndexes( Mutator<ByteBuffer> batch, String propertyName, Object propertyValue,
 +                                            boolean entitySchemaHasProperty, boolean noRead, UUID timestampUuid )
 +            throws Exception {
 +
 +        Entity entity = getHeadEntity();
 +
 +        UUID associatedId = null;
 +        String associatedType = null;
 +
 +        if ( Schema.isAssociatedEntityType( entity.getType() ) ) {
 +            Object item = entity.getProperty( PROPERTY_ITEM );
 +            if ( ( item instanceof UUID ) && ( entity.getProperty( PROPERTY_COLLECTION_NAME ) instanceof String ) ) {
 +                associatedId = ( UUID ) item;
 +                associatedType = string( entity.getProperty( PROPERTY_ITEM_TYPE ) );
 +                String entryName = TYPE_MEMBER + "." + propertyName;
 +                if ( logger.isDebugEnabled() ) {
 +                    logger.debug( "Extended property {} ( {} ).{} indexed as {} ({})." + entryName, new Object[] {
 +                            entity.getType(), entity.getUuid(), propertyName, associatedType, associatedId
 +                    } );
 +                }
 +                propertyName = entryName;
 +            }
 +        }
 +
 +        IndexUpdate indexUpdate = batchStartIndexUpdate( batch, entity, propertyName, propertyValue, timestampUuid,
 +                entitySchemaHasProperty, false, false,
 +                getDefaultSchema().isPropertyFulltextIndexed( entity.getType(), propertyName ), noRead );
 +
 +        // Update collections
 +
 +        String effectiveType = entity.getType();
 +        if ( associatedType != null ) {
 +            indexUpdate.setAssociatedId( associatedId );
 +            effectiveType = associatedType;
 +        }
 +
 +        Map<String, Set<CollectionInfo>> containers = getDefaultSchema().getContainers( effectiveType );
 +        if ( containers != null ) {
 +
 +            Map<EntityRef, Set<String>> containerEntities = null;
 +            if ( noRead ) {
 +                containerEntities = new LinkedHashMap<EntityRef, Set<String>>();
 +                EntityRef applicationRef = new SimpleEntityRef( TYPE_APPLICATION, applicationId );
 +                addMapSet( containerEntities, applicationRef, defaultCollectionName( entity.getType() ) );
 +            }
 +            else {
 +                containerEntities = getContainingCollections();
 +            }
 +
 +            for ( EntityRef containerEntity : containerEntities.keySet() ) {
 +                if ( containerEntity.getType().equals( TYPE_APPLICATION ) && Schema
 +                        .isAssociatedEntityType( entity.getType() ) ) {
 +                    logger.debug( "Extended properties for {} not indexed by application", entity.getType() );
 +                    continue;
 +                }
 +                Set<String> collectionNames = containerEntities.get( containerEntity );
 +                Set<CollectionInfo> collections = containers.get( containerEntity.getType() );
 +
 +                if ( collections != null ) {
 +                    for ( CollectionInfo collection : collections ) {
 +                        if ( collectionNames.contains( collection.getName() ) ) {
 +                            batchUpdateCollectionIndex( indexUpdate, containerEntity, collection.getName() );
 +                        }
 +                    }
 +                }
 +            }
 +        }
 +
 +        if ( !noRead ) {
 +            batchUpdateBackwardConnectionsPropertyIndexes( indexUpdate );
 +        }
 +
 +        /**
 +         * We've updated the properties, add the deletes to the ledger
 +         *
 +         */
 +
 +        for ( IndexEntry entry : indexUpdate.getPrevEntries() ) {
 +            addDeleteToMutator( batch, ENTITY_INDEX_ENTRIES, entity.getUuid(), entry.getLedgerColumn(),
 +                    indexUpdate.getTimestamp() );
 +        }
 +    }
 +
 +
 +    public void batchUpdateSetIndexes( Mutator<ByteBuffer> batch, String setName, Object elementValue,
 +                                       boolean removeFromSet, UUID timestampUuid ) throws Exception {
 +
 +        Entity entity = getHeadEntity();
 +
 +        elementValue = getDefaultSchema().validateEntitySetValue( entity.getType(), setName, elementValue );
 +
 +        IndexUpdate indexUpdate =
 +                batchStartIndexUpdate( batch, entity, setName, elementValue, timestampUuid, true, true, removeFromSet,
 +                        false );
 +
 +        // Update collections
 +        Map<String, Set<CollectionInfo>> containers =
 +                getDefaultSchema().getContainersIndexingDictionary( entity.getType(), setName );
 +
 +        if ( containers != null ) {
 +            Map<EntityRef, Set<String>> containerEntities = getContainingCollections();
 +            for ( EntityRef containerEntity : containerEntities.keySet() ) {
 +                if ( containerEntity.getType().equals( TYPE_APPLICATION ) && Schema
 +                        .isAssociatedEntityType( entity.getType() ) ) {
 +                    logger.debug( "Extended properties for {} not indexed by application", entity.getType() );
 +                    continue;
 +                }
 +                Set<String> collectionNames = containerEntities.get( containerEntity );
 +                Set<CollectionInfo> collections = containers.get( containerEntity.getType() );
 +
 +                if ( collections != null ) {
 +
 +                    for ( CollectionInfo collection : collections ) {
 +                        if ( collectionNames.contains( collection.getName() ) ) {
 +
 +                            batchUpdateCollectionIndex( indexUpdate, containerEntity, collection.getName() );
 +                        }
 +                    }
 +                }
 +            }
 +        }
 +
 +        batchUpdateBackwardConnectionsDictionaryIndexes( indexUpdate );
 +    }
 +
 +
 +    private IndexScanner searchIndex( Object indexKey, QuerySlice slice, int pageSize ) throws Exception {
 +
 +        DynamicComposite[] range = slice.getRange();
 +
 +        Object keyPrefix = key( indexKey, slice.getPropertyName() );
 +
 +        IndexScanner scanner =
 +                new IndexBucketScanner( cass, indexBucketLocator, ENTITY_INDEX, applicationId, IndexType.CONNECTION,
-                         keyPrefix, range[0], range[1], slice.isReversed(), pageSize, slice.getPropertyName() );
++                        keyPrefix, range[0], range[1], slice.isReversed(), pageSize, slice.hasCursor(), slice.getPropertyName() );
 +
 +        return scanner;
 +    }
 +
 +
 +    /**
 +     * Search the collection index using all the buckets for the given collection
 +     *
 +     * @param indexKey The index key to read
 +     * @param slice Slice set in the query
 +     * @param collectionName The name of the collection to search
 +     * @param pageSize The page size to load when iterating
 +     */
 +    private IndexScanner searchIndexBuckets( Object indexKey, QuerySlice slice, String collectionName, int pageSize )
 +            throws Exception {
 +
 +        DynamicComposite[] range = slice.getRange();
 +
 +        Object keyPrefix = key( indexKey, slice.getPropertyName() );
 +
-         // we have a cursor, so the first record should be discarded
-         if ( slice.hasCursor() ) {
-             pageSize++;
-         }
- 
 +        IndexScanner scanner =
 +                new IndexBucketScanner( cass, indexBucketLocator, ENTITY_INDEX, applicationId, IndexType.COLLECTION,
-                         keyPrefix, range[0], range[1], slice.isReversed(), pageSize, collectionName );
++                        keyPrefix, range[0], range[1], slice.isReversed(), pageSize, slice.hasCursor(), collectionName );
 +
 +        return scanner;
 +    }
 +
 +
 +    @SuppressWarnings("unchecked")
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_isOwner")
 +    public boolean isCollectionMember( String collectionName, EntityRef entity ) throws Exception {
 +
 +        Keyspace ko = cass.getApplicationKeyspace( applicationId );
 +
 +        ByteBuffer col = DynamicComposite
 +                .toByteBuffer( asList( this.headEntity.getType(), collectionName, headEntity.getUuid() ) );
 +
 +        HColumn<ByteBuffer, ByteBuffer> result = cass.getColumn( ko, ENTITY_COMPOSITE_DICTIONARIES,
 +                key( entity.getUuid(), Schema.DICTIONARY_CONTAINER_ENTITIES ), col, be, be );
 +
 +        return result != null;
 +    }
 +
 +
 +    /** @param connectionName The name of hte connection */
 +    public boolean isConnectionMember( String connectionName, EntityRef entity ) throws Exception {
 +        Keyspace ko = cass.getApplicationKeyspace( applicationId );
 +
 +        Object key = key( this.headEntity.getUuid(), DICTIONARY_CONNECTED_ENTITIES, connectionName );
 +
 +        DynamicComposite start = new DynamicComposite( entity.getUuid() );
 +
 +        List<HColumn<ByteBuffer, ByteBuffer>> cols =
 +                cass.getColumns( ko, ENTITY_COMPOSITE_DICTIONARIES, key, start, null, 1, false );
 +
 +        if ( cols == null || cols.size() == 0 ) {
 +            return false;
 +        }
 +
 +        UUID returnedUUID = ( UUID ) DynamicComposite.fromByteBuffer( cols.get( 0 ).getName() ).get( 0 );
 +
 +        return entity.getUuid().equals( returnedUUID );
 +
 +
 +        //    addDeleteToMutator(batch, ENTITY_COMPOSITE_DICTIONARIES,
 +        //        key(connection.getConnectedEntityId(), DICTIONARY_CONNECTING_ENTITIES,
 +        // connection.getConnectionType()),
 +        //        asList(connection.getConnectingEntityId(), connection.getConnectingEntityType()), timestamp);
 +        //
 +        //
 +        //    ConnectionRefImpl ref = new ConnectionRefImpl(this.headEntity, connectionName, entity);
 +        //
 +        //
 +        //
 +        //
 +        //
 +        //
 +        //    HColumn<String, UUID> col = cass.getColumn(ko, ENTITY_CONNECTIONS, ref.getUuid(),
 +        //        ConnectionRefImpl.CONNECTED_ENTITY_ID, se, ue);
 +        //
 +        //
 +        //    getConnectedEntities(this.headEntity, connectionName, )
 +        //
 +        //    return col != null && entity.getUuid().equals(col.getValue());
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_getOwners")
 +    public Map<String, Map<UUID, Set<String>>> getOwners() throws Exception {
 +        Map<EntityRef, Set<String>> containerEntities = getContainingCollections();
 +        Map<String, Map<UUID, Set<String>>> owners = new LinkedHashMap<String, Map<UUID, Set<String>>>();
 +
 +        for ( EntityRef owner : containerEntities.keySet() ) {
 +            Set<String> collections = containerEntities.get( owner );
 +            for ( String collection : collections ) {
 +                MapUtils.addMapMapSet( owners, owner.getType(), owner.getUuid(), collection );
 +            }
 +        }
 +
 +        return owners;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_getCollections")
 +    public Set<String> getCollections() throws Exception {
 +
 +        Map<String, CollectionInfo> collections = getDefaultSchema().getCollections( headEntity.getType() );
 +        if ( collections == null ) {
 +            return null;
 +        }
 +
 +        return collections.keySet();
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_getCollection_start_result")
 +    public Results getCollection( String collectionName, UUID startResult, int count, Results.Level resultsLevel,
 +                                  boolean reversed ) throws Exception {
 +        // changed intentionally to delegate to search so that behavior is
 +        // consistent across all index access.
 +
 +        // TODO T.N fix cursor parsing here so startResult can be used in this
 +        // context. Needs a bit of refactor
 +        // for accommodating cursor I/O USERGRID-1750. A bit hacky, but until a
 +        // furthur refactor this works.
 +
 +        Query query = new Query().withResultsLevel( resultsLevel ).withReversed( reversed ).withLimit( count )
 +                                 .withStartResult( startResult );
 +
 +        return searchCollection( collectionName, query );
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_getCollecitonForQuery")
 +    public Results getCollection( String collectionName, Query query, Results.Level resultsLevel ) throws Exception {
 +
 +        // changed intentionally to delegate to search so that behavior is
 +        // consistent across all index access.
 +
 +        return searchCollection( collectionName, query );
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_addToCollection")
 +    public Entity addToCollection( String collectionName, EntityRef itemRef ) throws Exception {
 +
 +        Entity itemEntity = em.get( itemRef );
 +
 +        if ( itemEntity == null ) {
 +            return null;
 +        }
 +
 +        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
 +        if ( ( collection != null ) && !collection.getType().equals( itemRef.getType() ) ) {
 +            return null;
 +        }
 +
 +        UUID timestampUuid = newTimeUUID();
 +        Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
 +
 +        batchAddToCollection( batch, collectionName, itemEntity, timestampUuid );
 +
 +        if ( collection.getLinkedCollection() != null ) {
 +            getRelationManager( itemEntity )
 +                    .batchAddToCollection( batch, collection.getLinkedCollection(), getHeadEntity(), timestampUuid );
 +        }
 +
 +        batchExecute( batch, CassandraService.RETRY_COUNT );
 +
 +        return itemEntity;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_addToCollections")
 +    public Entity addToCollections( List<EntityRef> owners, String collectionName ) throws Exception {
 +
 +        Entity itemEntity = getHeadEntity();
 +
 +        Map<String, List<UUID>> collectionsByType = new LinkedHashMap<String, List<UUID>>();
 +        for ( EntityRef owner : owners ) {
 +            MapUtils.addMapList( collectionsByType, owner.getType(), owner.getUuid() );
 +        }
 +
 +        UUID timestampUuid = newTimeUUID();
 +        Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
 +
 +        Schema schema = getDefaultSchema();
 +        for ( Entry<String, List<UUID>> entry : collectionsByType.entrySet() ) {
 +            CollectionInfo collection = schema.getCollection( entry.getKey(), collectionName );
 +            if ( ( collection != null ) && !collection.getType().equals( headEntity.getType() ) ) {
 +                continue;
 +            }
 +            batchAddToCollections( batch, entry.getKey(), entry.getValue(), collectionName, itemEntity, timestampUuid );
 +
 +            if ( collection.getLinkedCollection() != null ) {
 +                logger.error(
 +                        "Bulk add to collections used on a linked collection, linked connection will not be updated" );
 +            }
 +        }
 +
 +        batchExecute( batch, CassandraService.RETRY_COUNT );
 +
 +        return null;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_createItemInCollection")
 +    public Entity createItemInCollection( String collectionName, String itemType, Map<String, Object> properties )
 +            throws Exception {
 +
 +        if ( headEntity.getUuid().equals( applicationId ) ) {
 +            if ( itemType.equals( TYPE_ENTITY ) ) {
 +                itemType = singularize( collectionName );
 +            }
 +            if ( itemType.equals( TYPE_ROLE ) ) {
 +                Long inactivity = ( Long ) properties.get( PROPERTY_INACTIVITY );
 +                if ( inactivity == null ) {
 +                    inactivity = 0L;
 +                }
 +                return em.createRole( ( String ) properties.get( PROPERTY_NAME ),
 +                        ( String ) properties.get( PROPERTY_TITLE ), inactivity );
 +            }
 +            return em.create( itemType, properties );
 +        }
 +        else if ( headEntity.getType().equals( Group.ENTITY_TYPE ) && ( collectionName.equals( COLLECTION_ROLES ) ) ) {
 +            UUID groupId = headEntity.getUuid();
 +            String roleName = ( String ) properties.get( PROPERTY_NAME );
 +            return em.createGroupRole( groupId, roleName, ( Long ) properties.get( PROPERTY_INACTIVITY ) );
 +        }
 +
 +        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
 +        if ( ( collection != null ) && !collection.getType().equals( itemType ) ) {
 +            return null;
 +        }
 +
 +        properties = getDefaultSchema().cleanUpdatedProperties( itemType, properties, true );
 +
 +        Entity itemEntity = em.create( itemType, properties );
 +
 +        if ( itemEntity != null ) {
 +            UUID timestampUuid = newTimeUUID();
 +            Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
 +
 +            batchAddToCollection( batch, collectionName, itemEntity, timestampUuid );
 +
 +            if ( collection.getLinkedCollection() != null ) {
 +                getRelationManager( itemEntity )
 +                        .batchAddToCollection( batch, collection.getLinkedCollection(), getHeadEntity(),
 +                                timestampUuid );
 +            }
 +
 +            batchExecute( batch, CassandraService.RETRY_COUNT );
 +        }
 +
 +        return itemEntity;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_removeFromCollection")
 +    public void removeFromCollection( String collectionName, EntityRef itemRef ) throws Exception {
 +
 +        if ( headEntity.getUuid().equals( applicationId ) ) {
 +            if ( collectionName.equals( COLLECTION_ROLES ) ) {
 +                Entity itemEntity = em.get( itemRef );
 +                if ( itemEntity != null ) {
 +                    RoleRef roleRef = SimpleRoleRef.forRoleEntity( itemEntity );
 +                    em.deleteRole( roleRef.getApplicationRoleName() );
 +                    return;
 +                }
 +                em.delete( itemEntity );
 +                return;
 +            }
 +            em.delete( itemRef );
 +            return;
 +        }
 +
 +        Entity itemEntity = em.get( itemRef );
 +
 +        if ( itemEntity == null ) {
 +            return;
 +        }
 +
 +        UUID timestampUuid = newTimeUUID();
 +        Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
 +
 +        batchRemoveFromCollection( batch, collectionName, itemEntity, timestampUuid );
 +
 +        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
 +        if ( ( collection != null ) && ( collection.getLinkedCollection() != null ) ) {
 +            getRelationManager( itemEntity )
 +                    .batchRemoveFromCollection( batch, collection.getLinkedCollection(), getHeadEntity(),
 +                            timestampUuid );
 +        }
 +
 +        batchExecute( batch, CassandraService.RETRY_COUNT );
 +
 +        if ( headEntity.getType().equals( Group.ENTITY_TYPE ) ) {
 +            if ( collectionName.equals( COLLECTION_ROLES ) ) {
 +                String path = ( String ) ( ( Entity ) itemRef ).getMetadata( "path" );
 +                if ( path.startsWith( "/roles/" ) ) {
 +                    RoleRef roleRef = SimpleRoleRef.forRoleEntity( itemEntity );
 +                    em.deleteRole( roleRef.getApplicationRoleName() );
 +                }
 +            }
 +        }
 +    }
 +
 +
 +    @Metered(group = "core", name = "RelationManager_batchRemoveFromContainers")
 +    public void batchRemoveFromContainers( Mutator<ByteBuffer> m, UUID timestampUuid ) throws Exception {
 +        Entity entity = getHeadEntity();
 +        // find all the containing collections
 +        Map<EntityRef, Set<String>> containers = getContainingCollections();
 +        if ( containers != null ) {
 +            for ( Entry<EntityRef, Set<String>> container : containers.entrySet() ) {
 +                for ( String collectionName : container.getValue() ) {
 +                    getRelationManager( container.getKey() )
 +                            .batchRemoveFromCollection( m, collectionName, entity, true, timestampUuid );
 +                }
 +            }
 +        }
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_copyRelationships")
 +    public void copyRelationships( String srcRelationName, EntityRef dstEntityRef, String dstRelationName )
 +            throws Exception {
 +
 +        headEntity = em.validate( headEntity );
 +        dstEntityRef = em.validate( dstEntityRef );
 +
 +        CollectionInfo srcCollection = getDefaultSchema().getCollection( headEntity.getType(), srcRelationName );
 +
 +        CollectionInfo dstCollection = getDefaultSchema().getCollection( dstEntityRef.getType(), dstRelationName );
 +
 +        Results results = null;
 +        do {
 +            if ( srcCollection != null ) {
 +                results = em.getCollection( headEntity, srcRelationName, null, 5000, Level.REFS, false );
 +            }
 +            else {
 +                results = em.getConnectedEntities( headEntity.getUuid(), srcRelationName, null, Level.REFS );
 +            }
 +
 +            if ( ( results != null ) && ( results.size() > 0 ) ) {
 +                List<EntityRef> refs = results.getRefs();
 +                for ( EntityRef ref : refs ) {
 +                    if ( dstCollection != null ) {
 +                        em.addToCollection( dstEntityRef, dstRelationName, ref );
 +                    }
 +                    else {
 +                        em.createConnection( dstEntityRef, dstRelationName, ref );
 +                    }
 +                }
 +            }
 +        }
 +        while ( ( results != null ) && ( results.hasMoreResults() ) );
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_searchCollection")
 +    public Results searchCollection( String collectionName, Query query ) throws Exception {
 +
 +        if ( query == null ) {
 +            query = new Query();
 +        }
 +
 +        headEntity = em.validate( headEntity );
 +
 +        CollectionInfo collection = getDefaultSchema().getCollection( headEntity.getType(), collectionName );
 +
 +        query.setEntityType( collection.getType() );
 +
 +        final CollectionResultsLoaderFactory factory = new CollectionResultsLoaderFactory();
 +
 +        // we have something to search with, visit our tree and evaluate the
 +        // results
 +        QueryProcessor qp = new QueryProcessor( query, collection, em, factory );
 +        SearchCollectionVisitor visitor = new SearchCollectionVisitor( qp );
 +
 +        return qp.getResults( visitor );
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_createConnection_connection_ref")
 +    public ConnectionRef createConnection( ConnectionRef connection ) throws Exception {
 +        ConnectionRefImpl connectionImpl = new ConnectionRefImpl( connection );
 +
 +        updateEntityConnection( false, connectionImpl );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_createConnection_connectionType")
 +    public ConnectionRef createConnection( String connectionType, EntityRef connectedEntityRef ) throws Exception {
 +
 +        headEntity = em.validate( headEntity );
 +        connectedEntityRef = em.validate( connectedEntityRef );
 +
 +        ConnectionRefImpl connection = new ConnectionRefImpl( headEntity, connectionType, connectedEntityRef );
 +
 +        updateEntityConnection( false, connection );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_createConnection_paired_connection_type")
 +    public ConnectionRef createConnection( String pairedConnectionType, EntityRef pairedEntity, String connectionType,
 +                                           EntityRef connectedEntityRef ) throws Exception {
 +
 +        ConnectionRefImpl connection =
 +                new ConnectionRefImpl( headEntity, new ConnectedEntityRefImpl( pairedConnectionType, pairedEntity ),
 +                        new ConnectedEntityRefImpl( connectionType, connectedEntityRef ) );
 +
 +        updateEntityConnection( false, connection );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_createConnection_connected_entity_ref")
 +    public ConnectionRef createConnection( ConnectedEntityRef... connections ) throws Exception {
 +
 +        ConnectionRefImpl connection = new ConnectionRefImpl( headEntity, connections );
 +
 +        updateEntityConnection( false, connection );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_connectionRef_type_entity")
 +    public ConnectionRef connectionRef( String connectionType, EntityRef connectedEntityRef ) throws Exception {
 +
 +        ConnectionRef connection = new ConnectionRefImpl( headEntity, connectionType, connectedEntityRef );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_connectionRef_entity_to_entity")
 +    public ConnectionRef connectionRef( String pairedConnectionType, EntityRef pairedEntity, String connectionType,
 +                                        EntityRef connectedEntityRef ) throws Exception {
 +
 +        ConnectionRef connection =
 +                new ConnectionRefImpl( headEntity, new ConnectedEntityRefImpl( pairedConnectionType, pairedEntity ),
 +                        new ConnectedEntityRefImpl( connectionType, connectedEntityRef ) );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_connectionRef_connections")
 +    public ConnectionRef connectionRef( ConnectedEntityRef... connections ) {
 +
 +        ConnectionRef connection = new ConnectionRefImpl( headEntity, connections );
 +
 +        return connection;
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", name = "RelationManager_deleteConnection")
 +    public void deleteConnection( ConnectionRef connectionRef ) throws Exception {
 +        updateEntityConnection( true, new ConnectionRefImpl( connectionRef ) );
 +    }
 +
 +
 +    @Override
 +    @Metered(group = "core", 

<TRUNCATED>

Mime
View raw message