At 0 progress don't pretend to be 1 byte done.

----Release Notes----
At 0 progress don't pretend to be 1 byte done.

Until first record is done, progress should remain 0, otherwise wild interpolation of work estimate ensues.

Fraction consumed really should be named fraction_done or fraction_completed, because that is the progress signal required.

[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=121706839

Author: fahndrich

google/cloud/dataflow/io/range_trackers.py

@@ -101,7 +101,7 @@ def try_split_at_position(self, split_offset):
   @property
   def fraction_consumed(self):
     with self.lock:
-      fraction = ((1.0 * (self.last_record_start - self.start_position + 1) /
+      fraction = ((1.0 * (self.last_record_start - self.start_position) /
                    (self.stop_position - self.start_position)) if
                   self.stop_position != self.start_position else 0.0)
 

google/cloud/dataflow/io/range_trackers_test.py

@@ -110,25 +110,26 @@ def test_get_fraction_consumed_dense(self):
     tracker = range_trackers.OffsetRangeTracker(3, 6)
     self.assertEqual(0, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(True, 3))
-    self.assertEqual(1.0 / 3, tracker.fraction_consumed)
+    self.assertEqual(0.0, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(True, 4))
-    self.assertEqual(2.0 / 3, tracker.fraction_consumed)
+    self.assertEqual(1.0 / 3, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(True, 5))
-    self.assertEqual(1.0, tracker.fraction_consumed)
+    self.assertEqual(2.0 / 3, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(False, 6))  # non-split-point
+    self.assertEqual(1.0, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(False, 7))  # non-split-point
     self.assertFalse(tracker.try_return_record_at(True, 7))
 
   def test_get_fraction_consumed_sparse(self):
     tracker = range_trackers.OffsetRangeTracker(100, 200)
     self.assertEqual(0, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(True, 110))
-    # Consumed positions through 110 = total 11 positions of 100.
-    self.assertEqual(0.11, tracker.fraction_consumed)
+    # Consumed positions through 110 = total 10 positions of 100 done.
+    self.assertEqual(0.10, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(True, 150))
-    self.assertEqual(0.51, tracker.fraction_consumed)
+    self.assertEqual(0.50, tracker.fraction_consumed)
     self.assertTrue(tracker.try_return_record_at(True, 195))
-    self.assertEqual(0.96, tracker.fraction_consumed)
+    self.assertEqual(0.95, tracker.fraction_consumed)
 
   def test_everything_with_unbounded_range(self):
     tracker = range_trackers.OffsetRangeTracker(