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 2015/04/27 19:39:21 UTC
[12/20] incubator-usergrid git commit: Refactor into commands WIP
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionRefsVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionRefsVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionRefsVerifier.java
deleted file mode 100644
index b74f433..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionRefsVerifier.java
+++ /dev/null
@@ -1,44 +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.corepersistence.results;
-
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.usergrid.persistence.EntityRef;
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.SimpleEntityRef;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-
-public class CollectionRefsVerifier extends VersionVerifier {
-
-
-
- @Override
- public Results getResults( final Collection<Id> ids ) {
- List<EntityRef> refs = new ArrayList<EntityRef>(ids.size());
- for ( Id id : ids ) {
- refs.add( new SimpleEntityRef( id.getType(), id.getUuid() ) );
- }
- return Results.fromRefList( refs );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionResultsLoaderFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionResultsLoaderFactoryImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionResultsLoaderFactoryImpl.java
deleted file mode 100644
index 08458c3..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/CollectionResultsLoaderFactoryImpl.java
+++ /dev/null
@@ -1,59 +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.corepersistence.results;
-
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.SearchEdge;
-
-
-/**
- * Factory for creating results
- */
-public class CollectionResultsLoaderFactoryImpl implements ResultsLoaderFactory {
-
- private final ManagerCache managerCache;
-
-
- public CollectionResultsLoaderFactoryImpl( final ManagerCache managerCache ) {
- this.managerCache = managerCache;
- }
-
-
- @Override
- public ResultsLoader getLoader( final ApplicationScope applicationScope, final SearchEdge scope, final Query.Level resultsLevel ) {
-
- ResultsVerifier verifier;
-
- if ( resultsLevel == Query.Level.REFS ) {
- verifier = new CollectionRefsVerifier();
- }
- else if ( resultsLevel == Query.Level.IDS ) {
- verifier = new IdsVerifier();
- }
- else {
- verifier = new EntityVerifier(Query.MAX_LIMIT);
- }
-
- return new FilteringLoader( managerCache, verifier, applicationScope, scope );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionRefsVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionRefsVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionRefsVerifier.java
deleted file mode 100644
index 408edd3..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionRefsVerifier.java
+++ /dev/null
@@ -1,61 +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.corepersistence.results;
-
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-
-import org.apache.usergrid.persistence.ConnectedEntityRef;
-import org.apache.usergrid.persistence.ConnectionRef;
-import org.apache.usergrid.persistence.EntityRef;
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.cassandra.ConnectedEntityRefImpl;
-import org.apache.usergrid.persistence.cassandra.ConnectionRefImpl;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import static org.apache.usergrid.persistence.SimpleEntityRef.ref;
-
-
-/**
- * Verifier for creating connections
- */
-public class ConnectionRefsVerifier extends VersionVerifier {
-
-
- private final EntityRef ownerId;
- private final String connectionType;
-
-
- public ConnectionRefsVerifier( final EntityRef ownerId, final String connectionType ) {
- this.ownerId = ownerId;
- this.connectionType = connectionType;
- }
-
- @Override
- public Results getResults( final Collection<Id> ids ) {
- List<ConnectionRef> refs = new ArrayList<>();
- for ( Id id : ids ) {
- refs.add( new ConnectionRefImpl( ownerId, connectionType, ref(id.getType(), id.getUuid()) ));
- }
-
- return Results.fromConnections( refs );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionResultsLoaderFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionResultsLoaderFactoryImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionResultsLoaderFactoryImpl.java
deleted file mode 100644
index e9633c9..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ConnectionResultsLoaderFactoryImpl.java
+++ /dev/null
@@ -1,65 +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.corepersistence.results;
-
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.persistence.EntityRef;
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.SearchEdge;
-
-
-/**
- * Factory for creating results
- */
-public class ConnectionResultsLoaderFactoryImpl implements ResultsLoaderFactory {
-
- private final ManagerCache managerCache;
- private final EntityRef ownerId;
- private final String connectionType;
-
-
- public ConnectionResultsLoaderFactoryImpl( final ManagerCache managerCache, final EntityRef ownerId,
- final String connectionType ) {
- this.managerCache = managerCache;
- this.ownerId = ownerId;
- this.connectionType = connectionType;
- }
-
-
- @Override
- public ResultsLoader getLoader( final ApplicationScope applicationScope, final SearchEdge scope, final Query.Level resultsLevel ) {
-
- ResultsVerifier verifier;
-
- if ( resultsLevel == Query.Level.REFS ) {
- verifier = new ConnectionRefsVerifier( ownerId, connectionType );
- }
- else if ( resultsLevel == Query.Level.IDS ) {
- verifier = new ConnectionRefsVerifier( ownerId, connectionType );;
- }
- else {
- verifier = new EntityVerifier(Query.MAX_LIMIT);
- }
-
- return new FilteringLoader( managerCache, verifier, applicationScope, scope );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ElasticSearchQueryExecutor.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ElasticSearchQueryExecutor.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ElasticSearchQueryExecutor.java
deleted file mode 100644
index 4a1e43a..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ElasticSearchQueryExecutor.java
+++ /dev/null
@@ -1,240 +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.corepersistence.results;
-
-
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-
-import org.apache.usergrid.persistence.index.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-
-import com.google.common.base.Optional;
-
-
-public class ElasticSearchQueryExecutor implements QueryExecutor {
-
- private static final Logger logger = LoggerFactory.getLogger( ElasticSearchQueryExecutor.class );
-
- private final ResultsLoaderFactory resultsLoaderFactory;
-
- private final ApplicationScope applicationScope;
-
- private final ApplicationEntityIndex entityIndex;
-
- private final SearchEdge indexScope;
-
- private final SearchTypes types;
-
- private final Query query;
-
-
- private Results currentResults;
-
- private boolean moreToLoad = true;
-
-
- public ElasticSearchQueryExecutor( final ResultsLoaderFactory resultsLoaderFactory, final ApplicationEntityIndex entityIndex,
- final ApplicationScope applicationScope, final SearchEdge indexScope,
- final SearchTypes types, final Query query ) {
- this.resultsLoaderFactory = resultsLoaderFactory;
- this.applicationScope = applicationScope;
- this.entityIndex = entityIndex;
- this.indexScope = indexScope;
- this.types = types;
-
- //we must deep copy the query passed. Otherwise we will modify it's state with cursors. Won't fix, not relevant
- //once we start subscribing to streams.
- this.query = new Query(query);
- }
-
-
- @Override
- public Iterator<Results> iterator() {
- return this;
- }
-
-
- private void loadNextPage() {
- // Because of possible stale entities, which are filtered out by buildResults(),
- // we loop until the we've got enough results to satisfy the query limit.
-
- final int maxQueries = 10; // max re-queries to satisfy query limit
-
- final int originalLimit = query.getLimit();
-
- Results results = null;
- int queryCount = 0;
-
-
- CandidateResults crs = null;
-
- while ( queryCount++ < maxQueries ) {
-
- crs = getCandidateResults( query );
-
-
- logger.debug( "Calling build results with crs {}", crs );
- results = buildResults( indexScope, query, crs );
-
- /**
- * In an edge case where we delete stale entities, we could potentially get less results than expected.
- * This will only occur once during the repair phase.
- * We need to ensure that we short circuit before we overflow the requested limit during a repair.
- */
- if ( crs.isEmpty() || !crs.hasOffset() || results.size() > 0 ) { // no results, no cursor, can't get more
- break;
- }
-
-
- //we didn't load anything, but there was a cursor, this means a read repair occured. We have to short
- //circuit to avoid over returning the result set
-
-
- // need to query for more
- // ask for just what we need to satisfy, don't want to exceed limit
- query.setOffsetFromCursor(results.getCursor());
- query.setLimit( originalLimit - results.size() );
-
- logger.warn( "Satisfy query limit {}, new limit {} query count {}", new Object[] {
- originalLimit, query.getLimit(), queryCount
- } );
- }
-
- //now set our cursor if we have one for the next iteration
- if ( results.hasCursor() ) {
- query.setOffsetFromCursor(results.getCursor());
- moreToLoad = true;
- }
-
- else {
- moreToLoad = false;
- }
-
-
- //set our select subjects into our query if provided
- if(crs != null){
- query.setSelectSubjects( crs.getGetFieldMappings() );
- }
-
-
- //set our current results and the flag
- this.currentResults = results;
- }
-
-
- /**
- * Get the candidates or load the cursor, whichever we require
- * @param query
- * @return
- */
- private CandidateResults getCandidateResults(final Query query){
- final Optional<Integer> cursor = query.getOffset();
- final String queryToExecute = query.getQl().or("select *");
-
- CandidateResults results = cursor.isPresent()
- ? entityIndex.search( indexScope, types, queryToExecute, query.getLimit() , cursor.get())
- : entityIndex.search( indexScope, types, queryToExecute, query.getLimit());
-
- return results;
- }
-
-
- /**
- * Build results from a set of candidates, and discard those that represent stale indexes.
- *
- * @param query Query that was executed
- * @param crs Candidates to be considered for results
- */
- private Results buildResults( final SearchEdge indexScope, final Query query, final CandidateResults crs ) {
-
- logger.debug( "buildResults() from {} candidates", crs.size() );
-
- //get an instance of our results loader
- final ResultsLoader resultsLoader =
- this.resultsLoaderFactory.getLoader( applicationScope, indexScope, query.getResultsLevel() );
-
- //load the results
- final Results results = resultsLoader.loadResults(crs);
-
- //signal for post processing
- resultsLoader.postProcess();
-
- //set offset into query
- if(crs.getOffset().isPresent()) {
- query.setOffset(crs.getOffset().get());
- }else{
- query.clearOffset();
- }
- results.setCursorFromOffset( query.getOffset() );
-
- //ugly and tight coupling, but we don't have a choice until we finish some refactoring
- results.setQueryExecutor( this );
-
- logger.debug( "Returning results size {}", results.size() );
-
- return results;
- }
-
-
- @Override
- public boolean hasNext() {
-
- //we've tried to load and it's empty and we have more to load, load the next page
- if ( currentResults == null ) {
- //there's nothing left to load, nothing to do
- if ( !moreToLoad ) {
- return false;
- }
-
- //load the page
-
- loadNextPage();
- }
-
-
- //see if our current results are not null
- return currentResults != null;
- }
-
-
- @Override
- public Results next() {
- if ( !hasNext() ) {
- throw new NoSuchElementException( "No more results present" );
- }
-
- final Results toReturn = currentResults;
-
- currentResults = null;
-
- return toReturn;
- }
-
- @Override
- public void remove() {
- throw new RuntimeException("Remove not implemented!!");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
deleted file mode 100644
index a2dd5e0..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/EntityVerifier.java
+++ /dev/null
@@ -1,127 +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.corepersistence.results;
-
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.corepersistence.util.CpEntityMapUtils;
-import org.apache.usergrid.persistence.Entity;
-import org.apache.usergrid.persistence.EntityFactory;
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.collection.EntityCollectionManager;
-import org.apache.usergrid.persistence.collection.EntitySet;
-import org.apache.usergrid.persistence.collection.MvccEntity;
-import org.apache.usergrid.persistence.index.CandidateResult;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.fasterxml.uuid.UUIDComparator;
-import com.google.common.base.Optional;
-
-
-/**
- * A loader that verifies versions are correct in cassandra and match elasticsearch
- */
-public class EntityVerifier implements ResultsVerifier {
-
- private static final Logger logger = LoggerFactory.getLogger( EntityVerifier.class );
-
- private EntitySet ids;
-
- private Map<Id, org.apache.usergrid.persistence.model.entity.Entity> entityMapping;
-
-
- public EntityVerifier( final int maxSize ) {
- this.entityMapping = new HashMap<>( maxSize );
- }
-
-
- @Override
- public void loadResults( final Collection<Id> idsToLoad, final EntityCollectionManager ecm ) {
- ids = ecm.load( idsToLoad ).toBlocking().last();
- logger.debug("loadResults() asked for {} ids and got {}", idsToLoad.size(), ids.size());
- }
-
-
- @Override
- public boolean isValid( final CandidateResult candidateResult ) {
- final Id entityId = candidateResult.getId();
-
- final MvccEntity savedEntity = ids.getEntity( entityId );
-
- //version wasn't found deindex
- if ( savedEntity == null ) {
- logger.warn( "Version for Entity {}:{} not found", entityId.getType(), entityId.getUuid() );
- return false;
- }
-
- final UUID candidateVersion = candidateResult.getVersion();
- final UUID savedVersion = savedEntity.getVersion();
-
- if ( UUIDComparator.staticCompare( savedVersion, candidateVersion ) > 0 ) {
- logger.warn( "Stale version of Entity uuid:{} type:{}, stale v:{}, latest v:{}", new Object[] {
- entityId.getUuid(), entityId.getType(), candidateVersion, savedEntity
- } );
-
- return false;
- }
-
-
- final Optional<org.apache.usergrid.persistence.model.entity.Entity> entity = savedEntity.getEntity();
-
- if ( !entity.isPresent() ) {
- logger.warn( "Entity uuid:{} version v:{} is deleted but indexed, this is a bug ",
- entityId.getUuid(), savedEntity.getEntity() );
- return false;
- }
-
- entityMapping.put( entityId, entity.get() );
-
- return true;
- }
-
-
- @Override
- public Results getResults( final Collection<Id> ids ) {
-
- final List<Entity> ugEntities = new ArrayList<>( ids.size() );
-
- for ( final Id id : ids ) {
- final org.apache.usergrid.persistence.model.entity.Entity cpEntity = entityMapping.get( id );
-
- Entity entity = EntityFactory.newEntity( id.getUuid(), id.getType() );
-
- Map<String, Object> entityMap = CpEntityMapUtils.toMap( cpEntity );
- entity.addProperties( entityMap );
- ugEntities.add( entity );
- }
-
- return Results.fromEntities( ugEntities );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
deleted file mode 100644
index a48e1b8..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/FilteringLoader.java
+++ /dev/null
@@ -1,226 +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.corepersistence.results;
-
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.TreeMap;
-import java.util.UUID;
-
-import javax.annotation.Nullable;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.corepersistence.ManagerCache;
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.collection.EntityCollectionManager;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.ApplicationEntityIndex;
-import org.apache.usergrid.persistence.index.EntityIndexBatch;
-import org.apache.usergrid.persistence.index.SearchEdge;
-import org.apache.usergrid.persistence.index.CandidateResult;
-import org.apache.usergrid.persistence.index.CandidateResults;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.fasterxml.uuid.UUIDComparator;
-import com.google.common.base.Function;
-import com.google.common.collect.Collections2;
-
-
-public class FilteringLoader implements ResultsLoader {
-
- private static final Logger logger = LoggerFactory.getLogger( FilteringLoader.class );
-
- private final ManagerCache managerCache;
- private final ResultsVerifier resultsVerifier;
- private final ApplicationScope applicationScope;
- private final SearchEdge indexScope;
- private final EntityIndexBatch indexBatch;
-
-
- /**
- * Create an instance of a filter loader
- *
- * @param managerCache The manager cache to load
- * @param resultsVerifier The verifier to verify the candidate results
- * @param applicationScope The application scope to perform the load
- * @param indexScope The index scope used in the search
- */
- protected FilteringLoader( final ManagerCache managerCache, final ResultsVerifier resultsVerifier,
- final ApplicationScope applicationScope, final SearchEdge indexScope ) {
-
- this.managerCache = managerCache;
- this.resultsVerifier = resultsVerifier;
- this.applicationScope = applicationScope;
- this.indexScope = indexScope;
-
- final ApplicationEntityIndex index = managerCache.getEntityIndex( applicationScope );
-
- indexBatch = index.createBatch();
- }
-
-
- @Override
- public Results loadResults( final CandidateResults crs ) {
-
-
- if ( crs.size() == 0 ) {
- return new Results();
- }
-
-
- // For each entity, holds the index it appears in our candidates for keeping ordering correct
- final Map<Id, Integer> orderIndex = new HashMap<>( crs.size() );
-
- // Maps the entity ids to our candidates
- final Map<Id, CandidateResult> maxCandidateMapping = new HashMap<>( crs.size() );
-
-
- final Iterator<CandidateResult> iter = crs.iterator();
-
-
- // TODO, in this case we're "optimizing" due to the limitations of collection scope.
- // Perhaps we should change the API to just be an application, then an "owner" scope?
-
- // Go through the candidates and group them by scope for more efficient retrieval.
- // Also remove duplicates before we even make a network call
- for ( int i = 0; iter.hasNext(); i++ ) {
-
- final CandidateResult currentCandidate = iter.next();
-
- final Id entityId = currentCandidate.getId();
-
- //check if we've seen this candidate by id
- final CandidateResult previousMax = maxCandidateMapping.get( entityId );
-
- //its not been seen, save it
- if ( previousMax == null ) {
- maxCandidateMapping.put( entityId, currentCandidate );
- orderIndex.put( entityId, i );
- continue;
- }
-
- //we have seen it, compare them
-
- final UUID previousMaxVersion = previousMax.getVersion();
-
- final UUID currentVersion = currentCandidate.getVersion();
-
-
- final CandidateResult toRemove;
- final CandidateResult toKeep;
-
- //current is newer than previous. Remove previous and keep current
- if ( UUIDComparator.staticCompare( currentVersion, previousMaxVersion ) > 0 ) {
- toRemove = previousMax;
- toKeep = currentCandidate;
- }
- //previously seen value is newer than current. Remove the current and keep the previously seen value
- else {
- toRemove = currentCandidate;
- toKeep = previousMax;
- }
-
- //this is a newer version, we know we already have a stale entity, add it to be cleaned up
-
-
- //de-index it
- logger.warn( "Stale version of Entity uuid:{} type:{}, stale v:{}, latest v:{}", new Object[] {
- entityId.getUuid(), entityId.getType(), toRemove.getVersion(), toKeep.getVersion()
- } );
-
- //deindex this document, and remove the previous maxVersion
- //we have to deindex this from our ownerId, since this is what gave us the reference
- indexBatch.deindex( indexScope, toRemove );
-
-
- //TODO, fire the entity repair cleanup task here instead of de-indexing
-
- //replace the value with a more current version
- maxCandidateMapping.put( entityId, toKeep );
- orderIndex.put( entityId, i );
- }
-
-
- //now everything is ordered, and older versions are removed. Batch fetch versions to verify
- // existence and correct versions
-
- final TreeMap<Integer, Id> sortedResults = new TreeMap<>();
-
-
- final Collection<Id> idsToLoad =
- Collections2.transform( maxCandidateMapping.values(), new Function<CandidateResult, Id>() {
- @Nullable
- @Override
- public Id apply( @Nullable final CandidateResult input ) {
- //NOTE this is never null, we won't need to check
- return input.getId();
- }
- } );
-
-
- //now using the scope, load the collection
-
- // Get the collection scope and batch load all the versions. We put all entities in
- // app/app for easy retrieval/ unless persistence changes, we never want to read from
- // any scope other than the app, app, scope name scope
- // final CollectionScope collScope = new CollectionScopeImpl(
- // applicationScope.getApplication(), applicationScope.getApplication(), scopeName);
-
- final EntityCollectionManager ecm = managerCache.getEntityCollectionManager( applicationScope );
-
-
- //load the results into the loader for this scope for validation
- resultsVerifier.loadResults( idsToLoad, ecm );
-
- //now let the loader validate each candidate. For instance, the "max" in this candidate
- //could still be a stale result, so it needs validated
- for ( final Id requestedId : idsToLoad ) {
-
- final CandidateResult cr = maxCandidateMapping.get( requestedId );
-
- //ask the loader if this is valid, if not discard it and de-index it
- if ( !resultsVerifier.isValid( cr ) ) {
- indexBatch.deindex( indexScope, cr );
- continue;
- }
-
- //if we get here we're good, we need to add this to our results
- final int candidateIndex = orderIndex.get( requestedId );
-
- sortedResults.put( candidateIndex, requestedId );
- }
-
-
- // NOTE DO NOT execute the batch here.
- // It changes the results and we need consistent paging until we aggregate all results
- return resultsVerifier.getResults( sortedResults.values() );
- }
-
-
- @Override
- public void postProcess() {
- this.indexBatch.execute();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/IdsVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/IdsVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/IdsVerifier.java
deleted file mode 100644
index fae6eb1..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/IdsVerifier.java
+++ /dev/null
@@ -1,46 +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.corepersistence.results;
-
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.UUID;
-
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-
-public class IdsVerifier extends VersionVerifier {
-
- @Override
- public Results getResults( final Collection<Id> ids ) {
-
- final List<UUID> returnIds = new ArrayList<>( ids.size() );
-
- for ( final Id id : ids ) {
- returnIds.add( id.getUuid() );
- }
-
-
- return Results.fromIdList( returnIds );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ObservableQueryExecutor.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ObservableQueryExecutor.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ObservableQueryExecutor.java
new file mode 100644
index 0000000..382fb64
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ObservableQueryExecutor.java
@@ -0,0 +1,76 @@
+/*
+ * 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.corepersistence.results;
+
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+import org.apache.usergrid.persistence.Results;
+
+import rx.Observable;
+
+
+/**
+ * Our proxy to allow us to subscribe to observable results, then return them as an interator. A bridge for 2.0 -> 1.0
+ * code
+ */
+public class ObservableQueryExecutor implements QueryExecutor {
+
+ private final Observable<Results> resultsObservable;
+
+ public Iterator<Results> iterator;
+
+
+ public ObservableQueryExecutor( final Observable<Results> resultsObservable ) {
+ this.resultsObservable = resultsObservable;
+ }
+
+
+ @Override
+ public Iterator<Results> iterator() {
+ return this;
+ }
+
+
+ @Override
+ public boolean hasNext() {
+
+ if ( iterator == null ) {
+ iterator = resultsObservable.toBlocking().getIterator();
+ }
+
+
+ return iterator.hasNext();
+ }
+
+
+ @Override
+ public Results next() {
+ if ( !hasNext() ) {
+ throw new NoSuchElementException( "No more results present" );
+ }
+ final Results next = iterator.next();
+
+ next.setQueryExecutor( this );
+
+ return next;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoader.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoader.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoader.java
deleted file mode 100644
index e8a7bdf..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoader.java
+++ /dev/null
@@ -1,43 +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.corepersistence.results;
-
-
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.index.CandidateResults;
-
-
-/**
- * Interface for loading results
- */
-public interface ResultsLoader {
-
- /**
- * Using the candidate results, load our results. Should filter stale results
- * @param crs The candidate result set
- * @return Results. Null safe, but may be empty
- */
- public Results loadResults( final CandidateResults crs);
-
- /**
- * Post process the load operation
- */
- public void postProcess();
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoaderFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoaderFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoaderFactory.java
deleted file mode 100644
index 3399a35..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsLoaderFactory.java
+++ /dev/null
@@ -1,41 +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.corepersistence.results;
-
-
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.SearchEdge;
-
-
-/**
- * Factory for creating results
- */
-public interface ResultsLoaderFactory {
-
- /**
- * Get the loader for results
- * @param applicationScope The application scope used to load results
- * @param indexScope The index scope used in the search
- * @param
- */
- ResultsLoader getLoader( final ApplicationScope applicationScope, final SearchEdge indexScope,
- final Query.Level resultsLevel );
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsVerifier.java
deleted file mode 100644
index 46e0983..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/ResultsVerifier.java
+++ /dev/null
@@ -1,52 +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.corepersistence.results;
-
-
-import java.util.Collection;
-import org.apache.usergrid.persistence.Results;
-import org.apache.usergrid.persistence.collection.EntityCollectionManager;
-import org.apache.usergrid.persistence.index.CandidateResult;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-
-public interface ResultsVerifier {
-
- /**
- * Load all the candidate ides for verification
- * @param ids The Id's to load
- * @param ecm The entity collection manager
- */
- public void loadResults(Collection<Id> ids, EntityCollectionManager ecm);
-
- /**
- * Return true if the candidate result is a valid result that should be retained. If it should
- * not it should also be removed from the list of possible return values in this loader
- * @param candidateResult
- */
- public boolean isValid(CandidateResult candidateResult);
-
-
- /**
- * Load the result set with the given ids
- * @return
- */
- public Results getResults(Collection<Id> ids);
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d5978cca/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/VersionVerifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/VersionVerifier.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/VersionVerifier.java
deleted file mode 100644
index 574ba55..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/results/VersionVerifier.java
+++ /dev/null
@@ -1,85 +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.corepersistence.results;
-
-
-import java.util.Collection;
-import java.util.UUID;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.usergrid.persistence.collection.EntityCollectionManager;
-import org.apache.usergrid.persistence.collection.MvccLogEntry;
-import org.apache.usergrid.persistence.collection.VersionSet;
-import org.apache.usergrid.persistence.index.CandidateResult;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.fasterxml.uuid.UUIDComparator;
-
-
-/**
- * A loader that verifies versions are correct in Cassandra and match ElasticSearch
- */
-public abstract class VersionVerifier implements ResultsVerifier {
-
- private static final Logger logger = LoggerFactory.getLogger( VersionVerifier.class );
-
- private VersionSet ids;
-
-
- @Override
- public void loadResults( final Collection<Id> idsToLoad, final EntityCollectionManager ecm ) {
- ids = ecm.getLatestVersion( idsToLoad ).toBlocking().last();
- }
-
-
- @Override
- public boolean isValid( final CandidateResult candidateResult ) {
- final Id entityId = candidateResult.getId();
-
- final MvccLogEntry version = ids.getMaxVersion( entityId );
-
- //version wasn't found ,deindex
- if ( version == null ) {
- logger.warn( "Version for Entity {}:{} not found",
- entityId.getUuid(), entityId.getUuid() );
-
- return false;
- }
-
- final UUID savedVersion = version.getVersion();
-
- if ( UUIDComparator.staticCompare( savedVersion, candidateResult.getVersion() ) > 0 ) {
- logger.debug( "Stale version of Entity uuid:{} type:{}, stale v:{}, latest v:{}",
- new Object[] {
- entityId.getUuid(),
- entityId.getType(),
- candidateResult.getVersion(),
- savedVersion
- } );
-
- return false;
- }
-
- return true;
- }
-
-}