You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ja...@apache.org on 2015/04/22 19:10:58 UTC
[3/3] cassandra git commit: Fix write meter for cqlsh COPY TO
Fix write meter for cqlsh COPY TO
Patch by Tyler Hobbs; reviewed by Stefania Alborghetti for
CASSANDRA-9217
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5d504952
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5d504952
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5d504952
Branch: refs/heads/cassandra-2.1.5
Commit: 5d5049524b8087be9eaed6e146bba31e63083ae5
Parents: 533b470
Author: Tyler Hobbs <ty...@gmail.com>
Authored: Wed Apr 22 10:41:41 2015 -0500
Committer: T Jake Luciani <ja...@apache.org>
Committed: Wed Apr 22 13:08:26 2015 -0400
----------------------------------------------------------------------
bin/cqlsh | 57 ++++++++++++++---------
pylib/cqlshlib/test/test_cqlsh_completion.py | 20 ++++----
2 files changed, 45 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5d504952/bin/cqlsh
----------------------------------------------------------------------
diff --git a/bin/cqlsh b/bin/cqlsh
index e1bd312..f37fd49 100755
--- a/bin/cqlsh
+++ b/bin/cqlsh
@@ -1393,7 +1393,7 @@ class Shell(cmd.Cmd):
raise SyntaxError("Unknown direction %s" % direction)
timeend = time.time()
- print "%d rows %s in %s." % (rows, verb, describe_interval(timeend - timestart))
+ print "\n%d rows %s in %s." % (rows, verb, describe_interval(timeend - timestart))
def perform_csv_import(self, ks, cf, columns, fname, opts):
dialect_options = self.csv_dialect_defaults.copy()
@@ -1453,28 +1453,13 @@ class Shell(cmd.Cmd):
for process in processes:
process.start()
- last_checkpoint_time = time.time()
- current_rate = 0.0
+ meter = RateMeter(10000)
for current_record, row in enumerate(reader, start=1):
# write to the child process
pipes[current_record % num_processes].send((current_record, row))
# update the progress and current rate periodically
- if (current_record % 10000) == 0:
- new_checkpoint_time = time.time()
- new_rate = 10000.0 / (new_checkpoint_time - last_checkpoint_time)
- last_checkpoint_time = new_checkpoint_time
-
- # smooth the rate a bit
- if current_rate == 0.0:
- current_rate = new_rate
- else:
- current_rate = (current_rate + new_rate) / 2.0
-
- output = 'Processed %s rows; Write: %.2f rows/s\r' % \
- (current_record, current_rate)
- sys.stdout.write(output)
- sys.stdout.flush()
+ meter.increment()
# check for any errors reported by the children
if (current_record % 100) == 0:
@@ -1700,7 +1685,8 @@ class Shell(cmd.Cmd):
except IOError, e:
self.printerr("Can't open %r for writing: %s" % (fname, e))
return 0
- wmeter = meter.Meter()
+
+ meter = RateMeter(10000)
try:
dump = self.prep_export_dump(ks, cf, columns)
@@ -1713,12 +1699,11 @@ class Shell(cmd.Cmd):
time_format=self.display_time_format,
float_precision=self.display_float_precision).strval
writer.writerow(map(fmt, row.values()))
- wmeter.mark_written()
- wmeter.done()
+ meter.increment()
finally:
if do_close:
csvdest.close()
- return wmeter.num_finished()
+ return meter.current_record
def prep_export_dump(self, ks, cf, columns):
if columns is None:
@@ -2017,6 +2002,34 @@ class Shell(cmd.Cmd):
self.writeresult(text, color, newline=newline, out=sys.stderr)
+class RateMeter(object):
+
+ def __init__(self, log_rate):
+ self.log_rate = log_rate
+ self.last_checkpoint_time = time.time()
+ self.current_rate = 0.0
+ self.current_record = 0
+
+ def increment(self):
+ self.current_record += 1
+
+ if (self.current_record % self.log_rate) == 0:
+ new_checkpoint_time = time.time()
+ new_rate = self.log_rate / (new_checkpoint_time - self.last_checkpoint_time)
+ self.last_checkpoint_time = new_checkpoint_time
+
+ # smooth the rate a bit
+ if self.current_rate == 0.0:
+ self.current_rate = new_rate
+ else:
+ self.current_rate = (self.current_rate + new_rate) / 2.0
+
+ output = 'Processed %s rows; Write: %.2f rows/s\r' % \
+ (self.current_record, self.current_rate)
+ sys.stdout.write(output)
+ sys.stdout.flush()
+
+
class SwitchCommand(object):
command = None
description = None
http://git-wip-us.apache.org/repos/asf/cassandra/blob/5d504952/pylib/cqlshlib/test/test_cqlsh_completion.py
----------------------------------------------------------------------
diff --git a/pylib/cqlshlib/test/test_cqlsh_completion.py b/pylib/cqlshlib/test/test_cqlsh_completion.py
index 2d22a63..97bd96b 100644
--- a/pylib/cqlshlib/test/test_cqlsh_completion.py
+++ b/pylib/cqlshlib/test/test_cqlsh_completion.py
@@ -23,7 +23,7 @@ import re
from .basecase import BaseTestCase, cqlsh
from .cassconnect import testrun_cqlsh
-BEL = '\x07' # the terminal-bell character
+BEL = '\x07' # the terminal-bell character
CTRL_C = '\x03'
TAB = '\t'
@@ -35,6 +35,7 @@ COMPLETION_RESPONSE_TIME = 0.5
completion_separation_re = re.compile(r'\s+')
+
class CqlshCompletionCase(BaseTestCase):
def setUp(self):
self.cqlsh_runner = testrun_cqlsh(cqlver=cqlsh.DEFAULT_CQLVER, env={'COLUMNS': '100000'})
@@ -67,7 +68,7 @@ class CqlshCompletionCase(BaseTestCase):
if choice_output == BEL:
choice_output = ''
- self.cqlsh.send(CTRL_C) # cancel any current line
+ self.cqlsh.send(CTRL_C) # cancel any current line
self.cqlsh.read_to_next_prompt()
choice_lines = choice_output.splitlines()
@@ -127,12 +128,13 @@ class CqlshCompletionCase(BaseTestCase):
other_choices_ok=other_choices_ok,
split_completed_lines=split_completed_lines)
finally:
- self.cqlsh.send(CTRL_C) # cancel any current line
+ self.cqlsh.send(CTRL_C) # cancel any current line
self.cqlsh.read_to_next_prompt()
def strategies(self):
return self.module.CqlRuleSet.replication_strategies
+
class TestCqlshCompletion(CqlshCompletionCase):
cqlver = '3.1.6'
module = cqlsh.cql3handling
@@ -162,7 +164,6 @@ class TestCqlshCompletion(CqlshCompletionCase):
'users',
'has_all_types',
'system.',
- 'system_auth.',
'empty_composite_table',
'empty_table',
'undefined_values_table',
@@ -173,9 +174,9 @@ class TestCqlshCompletion(CqlshCompletionCase):
'songs'),
other_choices_ok=True)
self.trycompletions('INSERT INTO twenty_rows_composite_table',
- immediate=' ')
+ immediate=' (a, b ')
self.trycompletions('INSERT INTO twenty_rows_composite_table ',
- choices=['(', 'JSON'])
+ immediate='(a, b ')
self.trycompletions('INSERT INTO twenty_rows_composite_table (a, b ',
choices=(')', ','))
self.trycompletions('INSERT INTO twenty_rows_composite_table (a, b, ',
@@ -301,18 +302,17 @@ class TestCqlshCompletion(CqlshCompletionCase):
"VALUES ( 'eggs', 'sausage', 'spam') USING TTL 0 AND TIMESTAMP 0 AND "),
choices=[])
-
def test_complete_in_update(self):
self.trycompletions("UPD", immediate="ATE ")
self.trycompletions("UPDATE ",
- choices=['twenty_rows_table', 'system_auth.',
+ choices=['twenty_rows_table',
'users', 'has_all_types', 'system.',
'ascii_with_special_chars',
'empty_composite_table', 'empty_table',
'undefined_values_table',
'dynamic_columns',
'twenty_rows_composite_table',
- 'utf8_with_special_chars', 'ks.',
+ 'utf8_with_special_chars',
'system_traces.', 'songs'],
other_choices_ok=True)
@@ -346,7 +346,7 @@ class TestCqlshCompletion(CqlshCompletionCase):
self.trycompletions("UPDATE empty_table SET lonelycol = 'eggs' WHERE lonel",
choices=['<quotedName>', '<identifier>'])
self.trycompletions("UPDATE empty_table SET lonelycol = 'eggs' WHERE lonelykey ",
- choices=['=', '<=', '>=', '>', '<', 'CONTAINS', 'IN', '['])
+ choices=['=', '<=', '>=', '>', '<', 'CONTAINS', 'IN'])
self.trycompletions("UPDATE empty_table SET lonelycol = 'eggs' WHERE lonelykey = 0.0 ",
choices=['AND', 'IF', ';'])
self.trycompletions("UPDATE empty_table SET lonelycol = 'eggs' WHERE lonelykey = 0.0 AND ",