From 0bb3a064f3dd34823145124360c049f5d29b91ad Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Fri, 19 Oct 2012 17:59:09 -0500
Subject: [PATCH] fix progress counting in wide row iterator patch by Piotr
 Koalczkowski; reviewed by jbellis for CASSANDRA-4803

---
 .../hadoop/ColumnFamilyRecordReader.java      | 23 +++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
index fc90e5ccd4..73f978683a 100644
--- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
+++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
@@ -106,7 +106,9 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
 
     public float getProgress()
     {
-        // TODO this is totally broken for wide rows
+        if (!iter.hasNext())
+            return 1.0F;
+
         // the progress is likely to be reported slightly off the actual but close enough
         float progress = ((float) iter.rowsRead() / totalRowCount);
         return progress > 1.0F ? 1.0F : progress;
@@ -423,6 +425,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
     {
         private PeekingIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>>> wideColumns;
         private ByteBuffer lastColumn = ByteBufferUtil.EMPTY_BYTE_BUFFER;
+        private ByteBuffer lastCountedKey = ByteBufferUtil.EMPTY_BYTE_BUFFER;
 
         private void maybeInit()
         {
@@ -476,12 +479,28 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
             if (rows == null)
                 return endOfData();
 
-            totalRead++;
             Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> next = wideColumns.next();
             lastColumn = next.right.values().iterator().next().name();
+
+            maybeCountRow(next);
             return next;
         }
 
+
+        /**
+         * Increases the row counter only if we really moved to the next row.
+         * @param next just fetched row slice
+         */
+        private void maybeCountRow(Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>> next)
+        {
+            ByteBuffer currentKey = next.left;
+            if (!currentKey.equals(lastCountedKey))
+            {
+                totalRead++;
+                lastCountedKey = currentKey;
+            }
+        }
+
         private class WideColumnIterator extends AbstractIterator<Pair<ByteBuffer, SortedMap<ByteBuffer, IColumn>>>
         {
             private final Iterator<KeySlice> rows;