cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From slebre...@apache.org
Subject [7/17] git commit: fix KEYS index from skipping results patch by Dmitry Petrashko; reviewed by Pavel Yaskevich for CASSANDRA-3996
Date Tue, 13 Mar 2012 13:16:57 GMT
fix KEYS index from skipping results
patch by Dmitry Petrashko; reviewed by Pavel Yaskevich for CASSANDRA-3996


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/3db0a75d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/3db0a75d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/3db0a75d

Branch: refs/heads/trunk
Commit: 3db0a75d5bc971b219d08d0d514c97d6ac9c7aaf
Parents: c6f12ef
Author: Pavel Yaskevich <xedin@apache.org>
Authored: Tue Mar 13 14:10:13 2012 +0300
Committer: Pavel Yaskevich <xedin@apache.org>
Committed: Tue Mar 13 14:12:28 2012 +0300

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 +
 .../db/index/MultiRowIndexSearcherIterator.java    |  225 +++++++++++++++
 .../cassandra/db/index/keys/KeysSearcher.java      |  136 ++--------
 3 files changed, 247 insertions(+), 115 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/3db0a75d/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 5ed5299..3372a05 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,6 +1,7 @@
 1.1.1-dev
  * optimize commitlog checksumming (CASSANDRA-3610)
  * identify and blacklist corrupted SSTables from future compactions (CASSANDRA-2261)
+ * fix KEYS index from skipping results (CASSANDRA-3996)
 
 
 1.1-dev

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3db0a75d/src/java/org/apache/cassandra/db/index/MultiRowIndexSearcherIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/index/MultiRowIndexSearcherIterator.java b/src/java/org/apache/cassandra/db/index/MultiRowIndexSearcherIterator.java
new file mode 100644
index 0000000..ba7a023
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/index/MultiRowIndexSearcherIterator.java
@@ -0,0 +1,225 @@
+/**
+ * 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.cassandra.db.index;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.filter.ExtendedFilter;
+import org.apache.cassandra.db.filter.QueryFilter;
+import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.thrift.IndexExpression;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+
+/**
+ * This class is a general searcher that visits rows returned by nextIndexKey();
+ */
+public abstract class MultiRowIndexSearcherIterator extends ColumnFamilyStore.AbstractScanIterator
+{
+    private static final Logger logger = LoggerFactory.getLogger(MultiRowIndexSearcherIterator.class);
+    private static final Iterator<IColumn> EMPTY_ITERATOR = Collections.<IColumn>emptyList().iterator();
+
+    /* keys within a row */
+    protected final AbstractBounds<RowPosition> range;
+    private ByteBuffer lastSeenKey;
+    private final ByteBuffer startKey;
+    private final ByteBuffer endKey;
+
+    private Iterator<IColumn> currentIndexKeyData = null;
+    private final QueryPath path;
+
+    private final IndexExpression expression;
+    private final ExtendedFilter filter;
+    protected final ColumnFamilyStore indexCfs;
+    private final ColumnFamilyStore baseCfs;
+    private final boolean rightRangeIsNotMinimum;
+    protected DecoratedKey curIndexKey;
+
+    private final int rowsPerQuery;
+    private int columnsRead;
+
+
+    public MultiRowIndexSearcherIterator(IndexExpression expression,
+                                         ColumnFamilyStore baseCfs,
+                                         ColumnFamilyStore indexCfs,
+                                         ExtendedFilter filter,
+                                         AbstractBounds<RowPosition> range)
+    {
+        this.expression = expression;
+        this.baseCfs = baseCfs;
+        this.range = range;
+        this.filter = filter;
+        this.indexCfs = indexCfs;
+
+        /*
+        * XXX: If the range requested is a token range, we'll have to start at the beginning
(and stop at the end) of
+        * the indexed row unfortunately (which will be inefficient), because we have not
way to intuit the small
+        * possible key having a given token. A fix would be to actually store the token along
the key in the
+        * indexed row.
+        */
+        startKey = range.left instanceof DecoratedKey ? ((DecoratedKey) range.left).key :
ByteBufferUtil.EMPTY_BYTE_BUFFER;
+        endKey = range.right instanceof DecoratedKey ? ((DecoratedKey) range.right).key :
ByteBufferUtil.EMPTY_BYTE_BUFFER;
+
+        int meanColumns = Math.max(indexCfs.getMeanColumns(), 1);
+
+        // We shouldn't fetch only 1 row as this provides buggy paging in case the first
row doesn't satisfy all clauses
+        rowsPerQuery = Math.max(Math.min(filter.maxRows(), filter.maxColumns() / meanColumns),
2);
+        rightRangeIsNotMinimum = !range.right.isMinimum(baseCfs.partitioner);
+        path = new QueryPath(baseCfs.columnFamily);
+
+    }
+
+    /**
+     * This function should return indexCfs keys in order they would be scanned by searcher
+     * @return next key for scanning of null if endOfData
+     */
+    protected abstract DecoratedKey nextIndexKey();
+
+    /**
+     * resets internal state preparing for next indexCfs row scan.
+     */
+    protected void resetState()
+    {
+        curIndexKey = nextIndexKey();
+        currentIndexKeyData = EMPTY_ITERATOR;
+        lastSeenKey = startKey;
+        columnsRead = Integer.MAX_VALUE;
+    }
+
+    protected Row computeNext()
+    {
+        if (currentIndexKeyData == null) // this is first call. Initialize
+            resetState();
+
+        Row result = null;
+        while (result == null && curIndexKey != null) // curIndexKey would be null
when endOfData is reached
+        {
+            if (!currentIndexKeyData.hasNext()) // we've finished scanning row page
+            {
+                if (columnsRead < rowsPerQuery) // previously we've read less then we
queried. No more pages to read within this row
+                {
+                    logger.debug("Read only {} (< {}) last page through, must be done",
columnsRead, rowsPerQuery);
+                    resetState();
+                }
+                else
+                {
+                    if (logger.isDebugEnabled())
+                        logger.debug(String.format("Scanning index %s starting with %s",
+                                                   expressionString(expression), baseCfs.metadata.getKeyValidator().getString(startKey)));
+
+                    QueryFilter indexFilter = QueryFilter.getSliceFilter(curIndexKey,
+                                                                         new QueryPath(indexCfs.getColumnFamilyName()),
+                                                                         lastSeenKey,
+                                                                         endKey,
+                                                                         false,
+                                                                         rowsPerQuery);
+
+                    ColumnFamily indexRow = indexCfs.getColumnFamily(indexFilter); //get
next row page
+
+                    if (indexRow != null)
+                    {
+                        Collection<IColumn> sortedColumns = indexRow.getSortedColumns();
+                        columnsRead = sortedColumns.size();
+                        currentIndexKeyData = sortedColumns.iterator();
+                        IColumn firstColumn = sortedColumns.iterator().next();
+
+                        // Paging is racy, so it is possible the first column_name of a page
is not the last seen one.
+                        if (lastSeenKey != startKey && lastSeenKey.equals(firstColumn.name()))
+                        {
+                            // skip the row we already saw w/ the last page of results
+                            currentIndexKeyData.next();
+                            logger.debug("Skipping {}", baseCfs.metadata.getKeyValidator().getString(firstColumn.name()));
+                        }
+                        else if (range instanceof Range && currentIndexKeyData.hasNext()
&& firstColumn.name().equals(startKey))
+                        {
+                            // skip key excluded by range
+                            currentIndexKeyData.next();
+                            logger.debug("Skipping first key as range excludes it {}", baseCfs.metadata.getKeyValidator().getString(firstColumn.name()));
+                        }
+                    }
+                    else // page is empty, nothing to scan within this row
+                    {
+                        columnsRead = 0;
+                        currentIndexKeyData = EMPTY_ITERATOR;
+                    }
+                }
+            }
+
+
+            while (result == null && currentIndexKeyData.hasNext()) // rolling through
columns in page
+            {
+                IColumn column = currentIndexKeyData.next();
+                lastSeenKey = column.name();
+
+                if (column.isMarkedForDelete())
+                {
+                    logger.debug("Skipping {}", column);
+                    continue;
+                }
+
+                DecoratedKey dk = baseCfs.partitioner.decorateKey(lastSeenKey);
+
+                if (rightRangeIsNotMinimum && range.right.compareTo(dk) < 0) //
rightRangeIsNotMinimum is required to serve ring cycles
+                {
+                    logger.debug("Reached end of assigned scan range");
+                    resetState();
+                }
+                else if (range.contains(dk))
+                {
+                    logger.debug("Returning index hit for {}", dk);
+                    ColumnFamily data = baseCfs.getColumnFamily(new QueryFilter(dk, path,
filter.initialFilter()));
+
+                    // While the column family we'll get in the end should contains the primary
clause column_name,
+                    // the initialFilter may not have found it and can thus be null
+                    if (data == null)
+                        data = ColumnFamily.create(baseCfs.metadata);
+
+                    result = new Row(dk, data);
+                }
+                else
+                {
+                    logger.debug("Skipping entry {} outside of assigned scan range", dk.token);
+                }
+            }
+        }
+
+        return result == null ? endOfData() : result;
+    }
+
+    private String expressionString(IndexExpression expr)
+    {
+        return String.format("'%s.%s %s %s'",
+                             baseCfs.columnFamily,
+                             baseCfs.getComparator().getString(expr.column_name),
+                             expr.op,
+                             baseCfs.metadata.getColumn_metadata().get(expr.column_name).getValidator().getString(expr.value));
+    }
+
+    public void close() throws IOException
+    {}
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/3db0a75d/src/java/org/apache/cassandra/db/index/keys/KeysSearcher.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/index/keys/KeysSearcher.java b/src/java/org/apache/cassandra/db/index/keys/KeysSearcher.java
index bd4be7e..0914c64 100644
--- a/src/java/org/apache/cassandra/db/index/keys/KeysSearcher.java
+++ b/src/java/org/apache/cassandra/db/index/keys/KeysSearcher.java
@@ -17,20 +17,18 @@
  */
 package org.apache.cassandra.db.index.keys;
 
-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.*;
 
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.db.index.MultiRowIndexSearcherIterator;
 import org.apache.cassandra.db.index.SecondaryIndex;
 import org.apache.cassandra.db.index.SecondaryIndexManager;
 import org.apache.cassandra.db.index.SecondaryIndexSearcher;
 import org.apache.cassandra.dht.AbstractBounds;
-import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.thrift.IndexExpression;
 import org.apache.cassandra.thrift.IndexOperator;
-import org.apache.cassandra.utils.ByteBufferUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,15 +64,6 @@ public class KeysSearcher extends SecondaryIndexSearcher
         return best;
     }
 
-    private String expressionString(IndexExpression expr)
-    {
-        return String.format("'%s.%s %s %s'",
-                             baseCfs.columnFamily,
-                             baseCfs.getComparator().getString(expr.column_name),
-                             expr.op,
-                             baseCfs.metadata.getColumn_metadata().get(expr.column_name).getValidator().getString(expr.value));
-    }
-
     public boolean isIndexing(List<IndexExpression> clause)
     {
         return highestSelectivityPredicate(clause) != null;
@@ -95,115 +84,32 @@ public class KeysSearcher extends SecondaryIndexSearcher
         // TODO: allow merge join instead of just one index + loop
         final IndexExpression primary = highestSelectivityPredicate(filter.getClause());
         final SecondaryIndex index = indexManager.getIndexForColumn(primary.column_name);
+
         if (logger.isDebugEnabled())
             logger.debug("Primary scan clause is " + baseCfs.getComparator().getString(primary.column_name));
+
         assert index != null;
-        final DecoratedKey indexKey = indexManager.getIndexKeyFor(primary.column_name, primary.value);
+        return new KeysMultiRowIndexSearcherIterator(primary, filter, range, indexManager,
index);
+    }
 
-        /*
-         * XXX: If the range requested is a token range, we'll have to start at the beginning
(and stop at the end) of
-         * the indexed row unfortunately (which will be inefficient), because we have not
way to intuit the small
-         * possible key having a given token. A fix would be to actually store the token
along the key in the
-         * indexed row.
-         */
-        final ByteBuffer startKey = range.left instanceof DecoratedKey ? ((DecoratedKey)range.left).key
: ByteBufferUtil.EMPTY_BYTE_BUFFER;
-        final ByteBuffer endKey = range.right instanceof DecoratedKey ? ((DecoratedKey)range.right).key
: ByteBufferUtil.EMPTY_BYTE_BUFFER;
+    public class KeysMultiRowIndexSearcherIterator extends MultiRowIndexSearcherIterator
+    {
+        final DecoratedKey indexKey;
 
-        return new ColumnFamilyStore.AbstractScanIterator()
+        public KeysMultiRowIndexSearcherIterator(IndexExpression expression,
+                                                 ExtendedFilter filter,
+                                                 AbstractBounds<RowPosition> range,
+                                                 SecondaryIndexManager indexManager,
+                                                 SecondaryIndex index)
         {
-            private ByteBuffer lastSeenKey = startKey;
-            private Iterator<IColumn> indexColumns;
-            private final QueryPath path = new QueryPath(baseCfs.columnFamily);
-            private int columnsRead = Integer.MAX_VALUE;
-
-            protected Row computeNext()
-            {
-                int meanColumns = Math.max(index.getIndexCfs().getMeanColumns(), 1);
-                // We shouldn't fetch only 1 row as this provides buggy paging in case the
first row doesn't satisfy all clauses
-                int rowsPerQuery = Math.max(Math.min(filter.maxRows(), filter.maxColumns()
/ meanColumns), 2);
-                while (true)
-                {
-                    if (indexColumns == null || !indexColumns.hasNext())
-                    {
-                        if (columnsRead < rowsPerQuery)
-                        {
-                            logger.debug("Read only {} (< {}) last page through, must
be done", columnsRead, rowsPerQuery);
-                            return endOfData();
-                        }
-
-                        if (logger.isDebugEnabled())
-                            logger.debug(String.format("Scanning index %s starting with %s",
-                                                       expressionString(primary), index.getBaseCfs().metadata.getKeyValidator().getString(startKey)));
-
-                        QueryFilter indexFilter = QueryFilter.getSliceFilter(indexKey,
-                                                                             new QueryPath(index.getIndexCfs().getColumnFamilyName()),
-                                                                             lastSeenKey,
-                                                                             endKey,
-                                                                             false,
-                                                                             rowsPerQuery);
-                        ColumnFamily indexRow = index.getIndexCfs().getColumnFamily(indexFilter);
-                        logger.debug("fetched {}", indexRow);
-                        if (indexRow == null)
-                        {
-                            logger.debug("no data, all done");
-                            return endOfData();
-                        }
-
-                        Collection<IColumn> sortedColumns = indexRow.getSortedColumns();
-                        columnsRead = sortedColumns.size();
-                        indexColumns = sortedColumns.iterator();
-                        IColumn firstColumn = sortedColumns.iterator().next();
-
-                        // Paging is racy, so it is possible the first column of a page is
not the last seen one.
-                        if (lastSeenKey != startKey && lastSeenKey.equals(firstColumn.name()))
-                        {
-                            // skip the row we already saw w/ the last page of results
-                            indexColumns.next();
-                            columnsRead--;
-                            logger.debug("Skipping {}", baseCfs.metadata.getKeyValidator().getString(firstColumn.name()));
-                        }
-                        else if (range instanceof Range && indexColumns.hasNext()
&& firstColumn.name().equals(startKey))
-                        {
-                            // skip key excluded by range
-                            indexColumns.next();
-                            columnsRead--;
-                            logger.debug("Skipping first key as range excludes it");
-                        }
-                    }
-
-                    while (indexColumns.hasNext())
-                    {
-                        IColumn column = indexColumns.next();
-                        lastSeenKey = column.name();
-                        if (column.isMarkedForDelete())
-                        {
-                            logger.debug("skipping {}", column.name());
-                            continue;
-                        }
-
-                        DecoratedKey dk = baseCfs.partitioner.decorateKey(lastSeenKey);
-                        if (!range.right.isMinimum(baseCfs.partitioner) && range.right.compareTo(dk)
< 0)
-                        {
-                            logger.debug("Reached end of assigned scan range");
-                            return endOfData();
-                        }
-                        if (!range.contains(dk))
-                        {
-                            logger.debug("Skipping entry {} outside of assigned scan range",
dk.token);
-                            continue;
-                        }
-
-                        logger.debug("Returning index hit for {}", dk);
-                        ColumnFamily data = baseCfs.getColumnFamily(new QueryFilter(dk, path,
filter.initialFilter()));
-                        // While the column family we'll get in the end should contains the
primary clause column, the initialFilter may not have found it and can thus be null
-                        if (data == null)
-                            data = ColumnFamily.create(baseCfs.metadata);
-                        return new Row(dk, data);
-                    }
-                 }
-             }
+            super(expression, baseCfs, index.getIndexCfs(), filter, range);
+            indexKey = indexManager.getIndexKeyFor(expression.column_name, expression.value);
+        }
 
-            public void close() throws IOException {}
-        };
+        @Override
+        protected final DecoratedKey nextIndexKey()
+        {
+            return curIndexKey == null ? indexKey : null; // keys index always scan single
row in indexCfs
+        }
     }
 }


Mime
View raw message