You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sn...@apache.org on 2014/01/29 17:20:38 UTC
[4/9] git commit: Fixes off by 1 error at root iterators
Fixes off by 1 error at root iterators
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/8d4425d5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/8d4425d5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/8d4425d5
Branch: refs/heads/master
Commit: 8d4425d51ccf97d603dfeb0e587c432ad3c766a6
Parents: b7e3cf2
Author: Todd Nine <tn...@apigee.com>
Authored: Tue Jan 28 18:33:09 2014 -0700
Committer: Todd Nine <tn...@apigee.com>
Committed: Tue Jan 28 18:36:11 2014 -0700
----------------------------------------------------------------------
.../persistence/cassandra/CassandraService.java | 4 +-
.../cassandra/RelationManagerImpl.java | 21 ++----
.../cassandra/index/ConnectedIndexScanner.java | 77 ++++++++++++++++----
.../cassandra/index/IndexBucketScanner.java | 33 +++++++--
.../persistence/query/ir/SearchVisitor.java | 7 +-
.../query/ir/result/SliceIterator.java | 15 +---
.../query/AllInConnectionNoTypeIT.java | 1 +
.../ir/result/IntersectionIteratorTest.java | 6 +-
.../java/org/usergrid/tools/EntityCleanup.java | 2 +-
.../org/usergrid/tools/UniqueIndexCleanup.java | 2 +-
10 files changed, 108 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/main/java/org/usergrid/persistence/cassandra/CassandraService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/persistence/cassandra/CassandraService.java b/stack/core/src/main/java/org/usergrid/persistence/cassandra/CassandraService.java
index a6b6268..7edc530 100644
--- a/stack/core/src/main/java/org/usergrid/persistence/cassandra/CassandraService.java
+++ b/stack/core/src/main/java/org/usergrid/persistence/cassandra/CassandraService.java
@@ -1058,9 +1058,11 @@ public class CassandraService {
start = null;
}
+ final boolean skipFirst = start != null;
+
IndexScanner scanner =
new IndexBucketScanner( this, locator, ENTITY_ID_SETS, applicationId, IndexType.COLLECTION, key, start,
- finish, reversed, count, collectionName );
+ finish, reversed, count, skipFirst, collectionName );
return scanner;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/main/java/org/usergrid/persistence/cassandra/RelationManagerImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/persistence/cassandra/RelationManagerImpl.java b/stack/core/src/main/java/org/usergrid/persistence/cassandra/RelationManagerImpl.java
index 4f565c7..7eb7693 100644
--- a/stack/core/src/main/java/org/usergrid/persistence/cassandra/RelationManagerImpl.java
+++ b/stack/core/src/main/java/org/usergrid/persistence/cassandra/RelationManagerImpl.java
@@ -1379,7 +1379,7 @@ public class RelationManagerImpl implements RelationManager {
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;
}
@@ -1400,14 +1400,9 @@ public class RelationManagerImpl implements RelationManager {
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;
}
@@ -2128,14 +2123,12 @@ public class RelationManagerImpl implements RelationManager {
startId = UUID_PARSER.parse( slice.getCursor() ).getUUID();
}
- boolean skipFirst = node.isForceKeepFirst() ? false : slice.hasCursor();
-
IndexScanner indexScanner = cass.getIdList( cass.getApplicationKeyspace( applicationId ),
key( headEntity.getUuid(), DICTIONARY_COLLECTIONS, collectionName ), startId, null,
queryProcessor.getPageSizeHint( node ), query.isReversed(), indexBucketLocator, applicationId,
collectionName );
- this.results.push( new SliceIterator( slice, indexScanner, UUID_PARSER, skipFirst ) );
+ this.results.push( new SliceIterator( slice, indexScanner, UUID_PARSER ) );
}
@@ -2267,10 +2260,6 @@ public class RelationManagerImpl implements RelationManager {
start = slice.getCursor();
}
- // we'll discard the first match, increase the size
- if ( start != null ) {
- size++;
- }
boolean skipFirst = node.isForceKeepFirst() ? false : slice.hasCursor();
@@ -2316,9 +2305,9 @@ public class RelationManagerImpl implements RelationManager {
IndexScanner connectionScanner =
new ConnectedIndexScanner( cass, dictionaryType, applicationId, entityIdToUse, connectionTypes,
- start, slice.isReversed(), size );
+ start, slice.isReversed(), size, skipFirst );
- this.results.push( new SliceIterator( slice, connectionScanner, connectionParser, skipFirst ) );
+ this.results.push( new SliceIterator( slice, connectionScanner, connectionParser ) );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/ConnectedIndexScanner.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/ConnectedIndexScanner.java b/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/ConnectedIndexScanner.java
index 265dc4d..378f2bd 100644
--- a/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/ConnectedIndexScanner.java
+++ b/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/ConnectedIndexScanner.java
@@ -34,7 +34,9 @@ import static org.usergrid.persistence.cassandra.ApplicationCF.ENTITY_COMPOSITE_
import static org.usergrid.persistence.cassandra.CassandraPersistenceUtils.key;
-/** @author tnine */
+/**
+ * @author tnine
+ */
public class ConnectedIndexScanner implements IndexScanner {
private final CassandraService cass;
@@ -44,24 +46,35 @@ public class ConnectedIndexScanner implements IndexScanner {
private final String dictionaryType;
private final UUID entityId;
private final Iterator<String> connectionTypes;
+ private final boolean skipFirst;
- /** Pointer to our next start read */
+
+ /**
+ * Pointer to our next start read
+ */
private ByteBuffer start;
- /** Set to the original value to start scanning from */
+ /**
+ * Set to the original value to start scanning from
+ */
private ByteBuffer scanStart;
- /** Iterator for our results from the last page load */
+ /**
+ * Iterator for our results from the last page load
+ */
private LinkedHashSet<HColumn<ByteBuffer, ByteBuffer>> lastResults;
- /** True if our last load loaded a full page size. */
+ /**
+ * True if our last load loaded a full page size.
+ */
private boolean hasMore = true;
private String currentConnectionType;
public ConnectedIndexScanner( CassandraService cass, String dictionaryType, UUID applicationId, UUID entityId,
- Iterator<String> connectionTypes, ByteBuffer start, boolean reversed, int pageSize ) {
+ Iterator<String> connectionTypes, ByteBuffer start, boolean reversed, int pageSize,
+ boolean skipFirst ) {
Assert.notNull( entityId, "Entity id for row key construction must be specified when searching graph indexes" );
// create our start and end ranges
@@ -74,6 +87,7 @@ public class ConnectedIndexScanner implements IndexScanner {
this.pageSize = pageSize;
this.dictionaryType = dictionaryType;
this.connectionTypes = connectionTypes;
+ this.skipFirst = skipFirst;
if ( connectionTypes.hasNext() ) {
@@ -106,14 +120,32 @@ public class ConnectedIndexScanner implements IndexScanner {
return false;
}
+ boolean skipFirst = this.skipFirst && start == scanStart;
+
+ int totalSelectSize = pageSize + 1;
+
+ //we're discarding the first, so increase our total size by 1 since this value will be inclusive in the seek
+ if ( skipFirst ) {
+ totalSelectSize++;
+ }
+
lastResults = new LinkedHashSet<HColumn<ByteBuffer, ByteBuffer>>();
+
+ //cleanup columns for later logic
+ //pointer to the first col we load
+ HColumn<ByteBuffer, ByteBuffer> first = null;
+
+ //pointer to the last column we load
+ HColumn<ByteBuffer, ByteBuffer> last = null;
+
//go through each connection type until we exhaust the result sets
while ( currentConnectionType != null ) {
//only load a delta size to get this next page
- int selectSize = pageSize + 1 - lastResults.size();
+ int selectSize = totalSelectSize - lastResults.size();
+
Object key = key( entityId, dictionaryType, currentConnectionType );
@@ -122,17 +154,24 @@ public class ConnectedIndexScanner implements IndexScanner {
cass.getColumns( cass.getApplicationKeyspace( applicationId ), ENTITY_COMPOSITE_DICTIONARIES, key,
start, null, selectSize, reversed );
+ final int resultSize = results.size();
+
+ if(resultSize > 0){
+
+ last = results.get( resultSize -1 );
+
+ if(first == null ){
+ first = results.get( 0 );
+ }
+ }
+
lastResults.addAll( results );
+
// we loaded a full page, there might be more
- if ( results.size() == selectSize ) {
+ if ( resultSize == selectSize ) {
hasMore = true;
- // set the bytebuffer for the next pass
- start = results.get( results.size() - 1 ).getName();
-
- lastResults.remove( lastResults.size() - 1 );
-
//we've loaded a full page
break;
}
@@ -152,6 +191,16 @@ public class ConnectedIndexScanner implements IndexScanner {
}
}
+ //remove the first element, we need to skip it
+ if ( skipFirst && first != null) {
+ lastResults.remove( first );
+ }
+
+ if ( hasMore && last != null ) {
+ // set the bytebuffer for the next pass
+ start = last.getName();
+ lastResults.remove( last );
+ }
return lastResults != null && lastResults.size() > 0;
}
@@ -199,7 +248,7 @@ public class ConnectedIndexScanner implements IndexScanner {
* @see java.util.Iterator#next()
*/
@Override
- @Metered(group = "core", name = "IndexBucketScanner_load")
+ @Metered( group = "core", name = "IndexBucketScanner_load" )
public Set<HColumn<ByteBuffer, ByteBuffer>> next() {
Set<HColumn<ByteBuffer, ByteBuffer>> returnVal = lastResults;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/IndexBucketScanner.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/IndexBucketScanner.java b/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/IndexBucketScanner.java
index 1a0776a..5c234d9 100644
--- a/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/IndexBucketScanner.java
+++ b/stack/core/src/main/java/org/usergrid/persistence/cassandra/index/IndexBucketScanner.java
@@ -55,6 +55,7 @@ public class IndexBucketScanner implements IndexScanner {
private final int pageSize;
private final String[] indexPath;
private final IndexType indexType;
+ private final boolean skipFirst;
/** Pointer to our next start read */
private Object start;
@@ -69,9 +70,10 @@ public class IndexBucketScanner implements IndexScanner {
private boolean hasMore = true;
+
public IndexBucketScanner( CassandraService cass, IndexBucketLocator locator, ApplicationCF columnFamily,
UUID applicationId, IndexType indexType, Object keyPrefix, Object start, Object finish,
- boolean reversed, int pageSize, String... indexPath ) {
+ boolean reversed, int pageSize, boolean skipFirst, String... indexPath) {
this.cass = cass;
this.indexBucketLocator = locator;
this.applicationId = applicationId;
@@ -80,7 +82,10 @@ public class IndexBucketScanner implements IndexScanner {
this.start = start;
this.finish = finish;
this.reversed = reversed;
- this.pageSize = pageSize;
+ this.skipFirst = skipFirst;
+
+ //we always add 1 to the page size. This is because we pop the last column for the next page of results
+ this.pageSize = pageSize+1;
this.indexPath = indexPath;
this.indexType = indexType;
this.scanStart = start;
@@ -121,24 +126,40 @@ public class IndexBucketScanner implements IndexScanner {
//if we skip the first we need to set the load to page size +2, since we'll discard the first
//and start paging at the next entity, otherwise we'll just load the page size we need
- int selectSize = pageSize + 1;
+ int selectSize = pageSize;
+
+ //we purposefully use instance equality. If it's a pointer to the same value, we need to increase by 1
+ //since we'll be skipping the first value
+
+ final boolean firstPageSkipFirst = this.skipFirst && start == scanStart;
+
+ if(firstPageSkipFirst){
+ selectSize++;
+ }
TreeSet<HColumn<ByteBuffer, ByteBuffer>> resultsTree = IndexMultiBucketSetLoader
.load( cass, columnFamily, applicationId, cassKeys, start, finish, selectSize, reversed );
+ //remove the first element, it's from a cursor value and we don't want to retain it
+
+
// we loaded a full page, there might be more
if ( resultsTree.size() == selectSize ) {
hasMore = true;
- // set the bytebuffer for the next pass
- start = resultsTree.last().getName();
- resultsTree.remove( resultsTree.last() );
+ // set the bytebuffer for the next pass
+ start = resultsTree.pollLast().getName();
}
else {
hasMore = false;
}
+ //remove the first element since it needs to be skipped AFTER the size check. Otherwise it will fail
+ if ( firstPageSkipFirst ) {
+ resultsTree.pollFirst();
+ }
+
lastResults = resultsTree;
return lastResults != null && lastResults.size() > 0;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/main/java/org/usergrid/persistence/query/ir/SearchVisitor.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/persistence/query/ir/SearchVisitor.java b/stack/core/src/main/java/org/usergrid/persistence/query/ir/SearchVisitor.java
index 665ddd2..02e8386 100644
--- a/stack/core/src/main/java/org/usergrid/persistence/query/ir/SearchVisitor.java
+++ b/stack/core/src/main/java/org/usergrid/persistence/query/ir/SearchVisitor.java
@@ -170,8 +170,7 @@ public abstract class SearchVisitor implements NodeVisitor {
if ( subResults == null ) {
QuerySlice firstFieldSlice = new QuerySlice( slice.getPropertyName(), -1 );
subResults =
- new SliceIterator( slice, secondaryIndexScan( orderByNode, firstFieldSlice ), COLLECTION_PARSER,
- slice.hasCursor() );
+ new SliceIterator( slice, secondaryIndexScan( orderByNode, firstFieldSlice ), COLLECTION_PARSER );
}
orderIterator = new OrderByIterator( slice, orderByNode.getSecondarySorts(), subResults, em,
@@ -190,7 +189,7 @@ public abstract class SearchVisitor implements NodeVisitor {
scanner = secondaryIndexScan( orderByNode, slice );
}
- SliceIterator joinSlice = new SliceIterator( slice, scanner, COLLECTION_PARSER, slice.hasCursor() );
+ SliceIterator joinSlice = new SliceIterator( slice, scanner, COLLECTION_PARSER);
IntersectionIterator union = new IntersectionIterator( queryProcessor.getPageSizeHint( orderByNode ) );
union.addIterator( joinSlice );
@@ -221,7 +220,7 @@ public abstract class SearchVisitor implements NodeVisitor {
for ( QuerySlice slice : node.getAllSlices() ) {
IndexScanner scanner = secondaryIndexScan( node, slice );
- intersections.addIterator( new SliceIterator( slice, scanner, COLLECTION_PARSER, slice.hasCursor() ) );
+ intersections.addIterator( new SliceIterator( slice, scanner, COLLECTION_PARSER) );
}
results.push( intersections );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/main/java/org/usergrid/persistence/query/ir/result/SliceIterator.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/usergrid/persistence/query/ir/result/SliceIterator.java b/stack/core/src/main/java/org/usergrid/persistence/query/ir/result/SliceIterator.java
index 6922b9f..82f60c2 100644
--- a/stack/core/src/main/java/org/usergrid/persistence/query/ir/result/SliceIterator.java
+++ b/stack/core/src/main/java/org/usergrid/persistence/query/ir/result/SliceIterator.java
@@ -47,7 +47,6 @@ public class SliceIterator implements ResultIterator {
private final SliceParser parser;
private final IndexScanner scanner;
private final int pageSize;
- private final boolean skipFirst;
/**
* Pointer to the uuid set until it's returned
@@ -74,13 +73,11 @@ public class SliceIterator implements ResultIterator {
* @param scanner The scanner to use to read the cols
* @param slice The slice used in the scanner
* @param parser The parser for the scanner results
- * @param skipFirst True if the first record should be skipped, used with cursors
*/
- public SliceIterator( QuerySlice slice, IndexScanner scanner, SliceParser parser, boolean skipFirst ) {
+ public SliceIterator( QuerySlice slice, IndexScanner scanner, SliceParser parser ) {
this.slice = slice;
this.parser = parser;
this.scanner = scanner;
- this.skipFirst = skipFirst;
this.pageSize = scanner.getPageSize();
this.cols = new LinkedHashMap<UUID, ScanColumn>( this.pageSize );
this.parsedCols = new LinkedHashSet<ScanColumn>( this.pageSize );
@@ -122,13 +119,6 @@ public class SliceIterator implements ResultIterator {
cols.clear();
- /**
- * Skip the first value, it's from the previous cursor
- */
- if ( skipFirst && results.hasNext() ) {
- results.next();
- }
-
parsedCols.clear();
while ( results.hasNext() ) {
@@ -142,9 +132,6 @@ public class SliceIterator implements ResultIterator {
continue;
}
-
- logger.debug( "Parsed column with uuid '{}'", parsed.getUUID() );
-
last = parsed;
cols.put( parsed.getUUID(), parsed );
parsedCols.add( parsed );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/test/java/org/usergrid/persistence/query/AllInConnectionNoTypeIT.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/usergrid/persistence/query/AllInConnectionNoTypeIT.java b/stack/core/src/test/java/org/usergrid/persistence/query/AllInConnectionNoTypeIT.java
index 07d4c46..34f8cfe 100644
--- a/stack/core/src/test/java/org/usergrid/persistence/query/AllInConnectionNoTypeIT.java
+++ b/stack/core/src/test/java/org/usergrid/persistence/query/AllInConnectionNoTypeIT.java
@@ -24,6 +24,7 @@ import org.usergrid.persistence.Results;
/** @author tnine */
public class AllInConnectionNoTypeIT extends AbstractIteratingQueryIT {
+
@Test
public void allInConnectionNoType() throws Exception {
allIn( new ConnectionNoTypeHelper(app) );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/core/src/test/java/org/usergrid/persistence/query/ir/result/IntersectionIteratorTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/usergrid/persistence/query/ir/result/IntersectionIteratorTest.java b/stack/core/src/test/java/org/usergrid/persistence/query/ir/result/IntersectionIteratorTest.java
index 9ad4623..ac8c2cb 100644
--- a/stack/core/src/test/java/org/usergrid/persistence/query/ir/result/IntersectionIteratorTest.java
+++ b/stack/core/src/test/java/org/usergrid/persistence/query/ir/result/IntersectionIteratorTest.java
@@ -91,13 +91,13 @@ public class IntersectionIteratorTest {
// now make sure it's right, only 1, 3 and 8 intersect
assertTrue( union.hasNext() );
- assertEquals( id1, union.next().getUUID() );
+ assertEquals( id8, union.next().getUUID() );
assertTrue( union.hasNext() );
- assertEquals( id3, union.next().getUUID() );
+ assertEquals( id1, union.next().getUUID() );
assertTrue( union.hasNext() );
- assertEquals( id8, union.next().getUUID() );
+ assertEquals( id3, union.next().getUUID() );
assertFalse( union.hasNext() );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/tools/src/main/java/org/usergrid/tools/EntityCleanup.java
----------------------------------------------------------------------
diff --git a/stack/tools/src/main/java/org/usergrid/tools/EntityCleanup.java b/stack/tools/src/main/java/org/usergrid/tools/EntityCleanup.java
index ba79980..d886f70 100644
--- a/stack/tools/src/main/java/org/usergrid/tools/EntityCleanup.java
+++ b/stack/tools/src/main/java/org/usergrid/tools/EntityCleanup.java
@@ -126,7 +126,7 @@ public class EntityCleanup extends ToolBase {
key( applicationId, DICTIONARY_COLLECTIONS, collectionName ), null, null, PAGE_SIZE, false,
indexBucketLocator, applicationId, collectionName );
- SliceIterator itr = new SliceIterator( null, scanner, new UUIDIndexSliceParser(), false );
+ SliceIterator itr = new SliceIterator( null, scanner, new UUIDIndexSliceParser() );
while ( itr.hasNext() ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/8d4425d5/stack/tools/src/main/java/org/usergrid/tools/UniqueIndexCleanup.java
----------------------------------------------------------------------
diff --git a/stack/tools/src/main/java/org/usergrid/tools/UniqueIndexCleanup.java b/stack/tools/src/main/java/org/usergrid/tools/UniqueIndexCleanup.java
index 367d1da..f46a882 100644
--- a/stack/tools/src/main/java/org/usergrid/tools/UniqueIndexCleanup.java
+++ b/stack/tools/src/main/java/org/usergrid/tools/UniqueIndexCleanup.java
@@ -181,7 +181,7 @@ public class UniqueIndexCleanup extends ToolBase {
key( applicationId, DICTIONARY_COLLECTIONS, collectionName ), null, null, PAGE_SIZE, false,
indexBucketLocator, applicationId, collectionName );
- SliceIterator itr = new SliceIterator( null, scanner, new UUIDIndexSliceParser(), false );
+ SliceIterator itr = new SliceIterator( null, scanner, new UUIDIndexSliceParser() );
while ( itr.hasNext() ) {