cassandra-dtest-DEPRECATED/counter_tests.py at master · jmckenzie-dev/cassandra-dtest-DEPRECATED · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
from dtest import Tester
from assertions import *
from tools import *

import time

class TestCounters(Tester):

    def simple_increment_test(self):
        """ Simple incrementation test (Created for #3465, that wasn't a bug) """
        cluster = self.cluster

        cluster.populate(3).start()
        nodes = cluster.nodelist()

        cursor = self.patient_cql_connection(nodes[0]).cursor()
        self.create_ks(cursor, 'ks', 3)
        self.create_cf(cursor, 'cf', validation="CounterColumnType", columns={'c': 'counter'})
        cursor.close()


        cursors = [ self.patient_cql_connection(node, 'ks').cursor() for node in nodes ]
        nb_increment=50
        nb_counter=10

        for i in xrange(0, nb_increment):
            for c in xrange(0, nb_counter):
                cursor = cursors[(i + c) % len(nodes)]
                if cluster.version() >= '1.2':
                    cursor.execute("UPDATE cf SET c = c + 1 WHERE key = 'counter%i'" % c, consistency_level='QUORUM')
                else:
                    cursor.execute("UPDATE cf USING CONSISTENCY QUORUM SET c = c + 1 WHERE key = 'counter%i'" % c)

            cursor = cursors[i % len(nodes)]
            keys = ",".join(["'counter%i'" % c for c in xrange(0, nb_counter)])
            if cluster.version() >= '1.2':
                cursor.execute("SELECT key, c FROM cf WHERE key IN (%s)" % keys, consistency_level='QUORUM')
            else:
                cursor.execute("SELECT key, c FROM cf USING CONSISTENCY QUORUM WHERE key IN (%s)" % keys)
            res = cursor.fetchall()
            assert len(res) == nb_counter
            for c in xrange(0, nb_counter):
                assert len(res[c]) == 2, "Expecting key and counter for counter%i, got %s" % (c, str(res[c]))
                assert res[c][1] == i + 1, "Expecting counter%i = %i, got %i" % (c, i + 1, res[c][0])

    def upgrade_test(self):
        """ Test for bug of #4436 """

        cluster = self.cluster

        cluster.populate(2).start()
        nodes = cluster.nodelist()

        cql_version=None

        cursor = self.patient_cql_connection(nodes[0], version=cql_version).cursor()
        self.create_ks(cursor, 'ks', 2)

        query = """
            CREATE TABLE counterTable (
                k int PRIMARY KEY,
                c counter
            )
        """
        if cluster.version() >= '1.2':
            query = query +  "WITH compression = { 'sstable_compression' : 'SnappyCompressor' }"
        else:
            query = query +  "WITH compression_parameters:sstable_compression='SnappyCompressor'"

        cursor.execute(query)

        keys = range(0, 4)
        updates = 50

        def make_updates():
            cursor = self.patient_cql_connection(nodes[0], keyspace='ks', version=cql_version).cursor()
            upd = "UPDATE counterTable SET c = c + 1 WHERE k = %d;"
            #upd = "UPDATE counterTable SET c = c + 1 WHERE k = :k%d;"
            if cluster.version() >= '1.2':
                batch = " ".join(["BEGIN COUNTER BATCH"] + [upd % x for x in keys] + ["APPLY BATCH;"])
            else:
                batch = " ".join(["BEGIN BATCH USING CONSISTENCY LEVEL QUORUM"] + [upd % x for x in keys] + ["APPLY BATCH;"])

            #query = cursor.prepare_query(batch)

            kmap = { "k%d" % i : i for i in keys }
            for i in range(0, updates):
                if cluster.version() >= '1.2':
                    cursor.execute(batch, consistency_level='QUORUM')
                else:
                    cursor.execute(batch)

                #cursor.execute_prepared(query, kmap)

        def check(i):
            cursor = self.patient_cql_connection(nodes[0], keyspace='ks', version=cql_version).cursor()
            if cluster.version() >= '1.2':
                cursor.execute("SELECT * FROM counterTable", consistency_level='QUORUM')
            else:
                cursor.execute("SELECT * FROM counterTable USING CONSISTENCY QUORUM")
            assert cursor.rowcount == len(keys), "Expected %d rows, got %d: %s" % (len(keys), cursor.rowcount, str(cursor.fetchall()))
            for row in cursor:
                assert row[1] == i * updates, "Unexpected value %s" % str(row)

        def rolling_restart():
            # Rolling restart
            for i in range(0, 2):
                time.sleep(.2)
                nodes[i].nodetool("drain")
                nodes[i].stop(wait_other_notice=True)
                nodes[i].start(wait_other_notice=True)
                time.sleep(.2)

        make_updates()
        check(1)
        rolling_restart()

        make_updates()
        check(2)
        rolling_restart()

        make_updates()
        check(3)
        rolling_restart()

        check(3)