def sparse_supercolumn_test(self):
        cursor = self.prepare(row_factory=dict_factory)
        cluster = self.cluster

        node = self.cluster.nodelist()[0]
        node.nodetool("enablethrift")

        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)

        client.transport.open()
        client.set_keyspace('ks')

        cf = _create_sparse_super_cf('sparse_super_2')
        client.system_add_column_family(cf)

        for i in xrange(1, 3):
            client.insert('k1', ColumnParent('sparse_super_2', 'key{}'.format(i)), Column("value1", _i64(100), 0), ConsistencyLevel.ONE)
            client.insert('k1', ColumnParent('sparse_super_2', 'key{}'.format(i)), Column("col1", _i64(200), 0), ConsistencyLevel.ONE)
            client.insert('k1', ColumnParent('sparse_super_2', 'key{}'.format(i)), Column("col2", _i64(300), 0), ConsistencyLevel.ONE)

            client.insert('k2', ColumnParent('sparse_super_2', 'key{}'.format(i)), Column("value2", _i64(100), 0), ConsistencyLevel.ONE)
            client.insert('k2', ColumnParent('sparse_super_2', 'key{}'.format(i)), Column("col1", _i64(200), 0), ConsistencyLevel.ONE)
            client.insert('k2', ColumnParent('sparse_super_2', 'key{}'.format(i)), Column("col2", _i64(300), 0), ConsistencyLevel.ONE)

        _validate_sparse_thrift(client, cf='sparse_super_2')
        _validate_sparse_cql(cursor, cf='sparse_super_2')

        for is_upgraded, cursor in self.do_upgrade(cursor, row_factory=dict_factory, use_thrift=True):
            debug("Querying {} node".format("upgraded" if is_upgraded else "old"))
            client = get_thrift_client(host, port)
            _validate_sparse_thrift(client, cf='sparse_super_2')
            _validate_sparse_cql(cursor, cf='sparse_super_2')

    def dense_supercolumn_test_with_renames(self):
        cursor = self.prepare(row_factory=dict_factory)
        cluster = self.cluster

        node = self.cluster.nodelist()[0]
        node.nodetool("enablethrift")

        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)

        client.transport.open()
        client.set_keyspace('ks')

        client.system_add_column_family(_create_dense_super_cf('dense_super_2'))

        for i in xrange(1, 3):
            client.insert('k1', ColumnParent('dense_super_2', 'key{}'.format(i)), Column(_i64(100), 'value1', 0), ConsistencyLevel.ONE)
            client.insert('k2', ColumnParent('dense_super_2', 'key{}'.format(i)), Column(_i64(200), 'value2', 0), ConsistencyLevel.ONE)

        cursor.execute("ALTER TABLE ks.dense_super_2 RENAME key TO renamed_key")
        cursor.execute("ALTER TABLE ks.dense_super_2 RENAME column1 TO renamed_column1")
        cursor.execute("ALTER TABLE ks.dense_super_2 RENAME column2 TO renamed_column2")
        cursor.execute("ALTER TABLE ks.dense_super_2 RENAME value TO renamed_value")

        _validate_dense_cql(cursor, cf='dense_super_2', key=u'renamed_key', column1=u'renamed_column1', column2=u'renamed_column2', value=u'renamed_value')
        _validate_dense_thrift(client, cf='dense_super_2')

        for is_upgraded, cursor in self.do_upgrade(cursor, row_factory=dict_factory, use_thrift=True):
            debug("Querying {} node".format("upgraded" if is_upgraded else "old"))
            client = get_thrift_client(host, port)
            _validate_dense_cql(cursor, cf='dense_super_2', key=u'renamed_key', column1=u'renamed_column1', column2=u'renamed_column2', value=u'renamed_value')
            _validate_dense_thrift(client, cf='dense_super_2')

    def dense_supercolumn_3_0_created_test(self):
        cluster = self.prepare(cassandra_version='github:apache/cassandra-3.0')
        node = self.cluster.nodelist()[0]
        cursor = self.patient_cql_connection(node, row_factory=dict_factory)

        cursor.execute("CREATE KEYSPACE ks WITH replication = {'class': 'SimpleStrategy','replication_factor': '1' };")

        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)

        client.transport.open()
        client.set_keyspace('ks')

        client.system_add_column_family(_create_dense_super_cf('dense_super_1'))

        for i in xrange(1, 3):
            client.insert('k1', ColumnParent('dense_super_1', 'key{}'.format(i)), Column(_i64(100), 'value1', 0), ConsistencyLevel.ONE)
            client.insert('k2', ColumnParent('dense_super_1', 'key{}'.format(i)), Column(_i64(200), 'value2', 0), ConsistencyLevel.ONE)

        _validate_dense_thrift(client, cf='dense_super_1')

        node.stop()
        self.set_node_to_current_version(node)
        node.set_configuration_options(values={'start_rpc': 'true'})
        node.start()

        cursor = self.patient_cql_connection(node, row_factory=dict_factory)
        client = get_thrift_client(host, port)

        _validate_dense_thrift(client, cf='dense_super_1')
        _validate_dense_cql(cursor, cf='dense_super_1')

    def functional_test(self):
        NUM_SUBCOLS = 100
        NUM_ADDS = 100

        cluster = self.cluster
        cluster.populate(3).start()
        node1 = cluster.nodelist()[0]

        time.sleep(0.5)
        session = self.patient_cql_connection(node1)
        create_ks(session, "ks", 3)
        time.sleep(1)  # wait for propagation

        # create the columnfamily using thrift
        host, port = node1.network_interfaces["thrift"]
        thrift_conn = get_thrift_client(host, port)
        thrift_conn.transport.open()
        thrift_conn.set_keyspace("ks")
        cf_def = CfDef(keyspace="ks", name="cf", column_type="Super", default_validation_class="CounterColumnType")
        thrift_conn.system_add_column_family(cf_def)

        # let the sediment settle to to the bottom before drinking...
        time.sleep(2)

        for subcol in xrange(NUM_SUBCOLS):
            for add in xrange(NUM_ADDS):
                column_parent = ColumnParent(column_family="cf", super_column="subcol_%d" % subcol)
                counter_column = CounterColumn("col_0", 1)
                thrift_conn.add("row_0", column_parent, counter_column, ConsistencyLevel.QUORUM)
        time.sleep(1)
        cluster.flush()

        debug("Stopping cluster")
        cluster.stop()
        time.sleep(5)
        debug("Starting cluster")
        cluster.start()
        time.sleep(5)

        thrift_conn = get_thrift_client(host, port)
        thrift_conn.transport.open()
        thrift_conn.set_keyspace("ks")

        from_db = []

        for i in xrange(NUM_SUBCOLS):
            column_path = ColumnPath(column_family="cf", column="col_0", super_column="subcol_%d" % i)
            column_or_super_column = thrift_conn.get("row_0", column_path, ConsistencyLevel.QUORUM)
            val = column_or_super_column.counter_column.value
            debug(str(val)),
            from_db.append(val)
        debug("")

        expected = [NUM_ADDS for i in xrange(NUM_SUBCOLS)]

        if from_db != expected:
            raise Exception("Expected a bunch of the same values out of the db. Got this: " + str(from_db))

    def rename_test(self):
        """
        Check that a thrift-created table can be renamed via CQL:

        - create a table via the thrift interface
        - INSERT a row via CQL
        - ALTER the name of the table via CQL
        - SELECT from the table and assert the values inserted are there
        """
        session = self.prepare(start_rpc=True)

        node = self.cluster.nodelist()[0]
        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()

        cfdef = CfDef()
        cfdef.keyspace = 'ks'
        cfdef.name = 'test'
        cfdef.column_type = 'Standard'
        cfdef.comparator_type = 'CompositeType(Int32Type, Int32Type, Int32Type)'
        cfdef.key_validation_class = 'UTF8Type'
        cfdef.default_validation_class = 'UTF8Type'

        client.set_keyspace('ks')
        client.system_add_column_family(cfdef)

        session.execute("INSERT INTO ks.test (key, column1, column2, column3, value) VALUES ('foo', 4, 3, 2, 'bar')")
        session.execute("ALTER TABLE test RENAME column1 TO foo1 AND column2 TO foo2 AND column3 TO foo3")
        assert_one(session, "SELECT foo1, foo2, foo3 FROM test", [4, 3, 2])

    def rename_test(self):
        session = self.prepare(start_rpc=True)

        node = self.cluster.nodelist()[0]
        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()

        cfdef = CfDef()
        cfdef.keyspace = 'ks'
        cfdef.name = 'test'
        cfdef.column_type = 'Standard'
        cfdef.comparator_type = 'CompositeType(Int32Type, Int32Type, Int32Type)'
        cfdef.key_validation_class = 'UTF8Type'
        cfdef.default_validation_class = 'UTF8Type'

        client.set_keyspace('ks')
        client.system_add_column_family(cfdef)

        session.execute("INSERT INTO ks.test (key, column1, column2, column3, value) VALUES ('foo', 4, 3, 2, 'bar')")

        time.sleep(1)

        session.execute("ALTER TABLE test RENAME column1 TO foo1 AND column2 TO foo2 AND column3 TO foo3")
        assert_one(session, "SELECT foo1, foo2, foo3 FROM test", [4, 3, 2])

    def cql3_insert_thrift_test(self):
        """ Check that we can insert from thrift into a CQL3 table (#4377) """
        session = self.prepare(start_rpc=True)

        session.execute("""
            CREATE TABLE test (
                k int,
                c int,
                v int,
                PRIMARY KEY (k, c)
            )
        """)

        node = self.cluster.nodelist()[0]
        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()
        client.set_keyspace('ks')
        key = struct.pack('>i', 2)
        column_name_component = struct.pack('>i', 4)
        # component length + component + EOC + component length + component + EOC
        column_name = '\x00\x04' + column_name_component + '\x00' + '\x00\x01' + 'v' + '\x00'
        value = struct.pack('>i', 8)
        client.batch_mutate(
            {key: {'test': [Mutation(ColumnOrSuperColumn(column=Column(name=column_name, value=value, timestamp=100)))]}},
            ThriftConsistencyLevel.ONE)

        res = session.execute("SELECT * FROM test")
        assert rows_to_list(res) == [[2, 4, 8]], res

    def upgrade_with_range_tombstone_eoc_0_test(self):
        """
        Check sstable upgrading when the sstable contains a range tombstone with EOC=0.

        @jira_ticket CASSANDRA-12423
        """
        session = self._setup_cluster(cluster_options={'start_rpc': 'true'})

        session.execute("CREATE TABLE rt (id INT, c1 TEXT, c2 TEXT, v INT, PRIMARY KEY (id, c1, c2)) "
                        "with compact storage and compression = {'sstable_compression': ''};")

        range_delete = {
            i32(1): {
                'rt': [Mutation(deletion=Deletion(2470761440040513,
                                                  predicate=SlicePredicate(slice_range=SliceRange(
                                                      start=composite('a', eoc='\x00'),
                                                      finish=composite('asd', eoc='\x00')))))]
            }
        }

        client = get_thrift_client()
        client.transport.open()
        client.set_keyspace('ks')
        client.batch_mutate(range_delete, ConsistencyLevel.ONE)
        client.transport.close()

        session.execute("INSERT INTO rt (id, c1, c2, v) VALUES (1, 'asd', '', 0) USING TIMESTAMP 1470761451368658")
        session.execute("INSERT INTO rt (id, c1, c2, v) VALUES (1, 'asd', 'asd', 0) USING TIMESTAMP 1470761449416613")

        session = self._do_upgrade()

        ret = list(session.execute('SELECT * FROM rt'))
        assert_length_equal(ret, 2)

    def test_thrift(self):
        """
        A thrift client receives a TimedOutException
        """
        self._prepare_cluster(start_rpc=True)
        self.expected_expt = thrift_types.TimedOutException

        client = get_thrift_client()
        client.transport.open()
        client.set_keyspace(KEYSPACE)

        with self.assertRaises(self.expected_expt):
            client.insert('key1',
                          thrift_types.ColumnParent('mytable'),
                          thrift_types.Column('value', 'Value 1', 0),
                          thrift_types.ConsistencyLevel.ALL)

        client.transport.close()

    def cql3_insert_thrift_test(self):
        """
        Check that we can insert from thrift into a CQL3 table:

        - CREATE a table via CQL
        - insert values via thrift
        - SELECT the inserted values and assert they are there as expected

        @jira_ticket CASSANDRA-4377
        """
        session = self.prepare(start_rpc=True)

        session.execute("""
            CREATE TABLE test (
                k int,
                c int,
                v int,
                PRIMARY KEY (k, c)
            )
        """)

        node = self.cluster.nodelist()[0]
        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()
        client.set_keyspace('ks')
        key = struct.pack('>i', 2)
        column_name_component = struct.pack('>i', 4)
        # component length + component + EOC + component length + component + EOC
        column_name = '\x00\x04' + column_name_component + '\x00' + '\x00\x01' + 'v' + '\x00'
        value = struct.pack('>i', 8)
        client.batch_mutate(
            {key: {'test': [Mutation(ColumnOrSuperColumn(column=Column(name=column_name, value=value, timestamp=100)))]}},
            ThriftConsistencyLevel.ONE)

        assert_one(session, "SELECT * FROM test", [2, 4, 8])

    def sc_with_row_cache_test(self):
        """ Test for bug reported in #4190 """
        cluster = self.cluster

        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]
        self.patient_cql_connection(node1)

        node = self.cluster.nodelist()[0]
        host, port = node.network_interfaces["thrift"]
        client = get_thrift_client(host, port)
        client.transport.open()

        ksdef = KsDef()
        ksdef.name = "ks"
        ksdef.strategy_class = "SimpleStrategy"
        ksdef.strategy_options = {"replication_factor": "1"}
        ksdef.cf_defs = []

        client.system_add_keyspace(ksdef)
        client.set_keyspace("ks")

        # create a super column family with UTF8 for all types
        cfdef = CfDef()
        cfdef.keyspace = "ks"
        cfdef.name = "Users"
        cfdef.column_type = "Super"
        cfdef.comparator_type = "UTF8Type"
        cfdef.subcomparator_type = "UTF8Type"
        cfdef.key_validation_class = "UTF8Type"
        cfdef.default_validation_class = "UTF8Type"
        cfdef.caching = "rows_only"

        client.system_add_column_family(cfdef)

        column = Column(name="name", value="Mina", timestamp=100)
        client.batch_mutate(
            {"mina": {"Users": [Mutation(ColumnOrSuperColumn(super_column=SuperColumn("attrs", [column])))]}},
            ThriftConsistencyLevel.ONE,
        )

        column_parent = ColumnParent(column_family="Users")
        predicate = SlicePredicate(slice_range=SliceRange("", "", False, 100))
        super_columns = client.get_slice("mina", column_parent, predicate, ThriftConsistencyLevel.ONE)
        self.assertEqual(1, len(super_columns))
        super_column = super_columns[0].super_column
        self.assertEqual("attrs", super_column.name)
        self.assertEqual(1, len(super_column.columns))
        self.assertEqual("name", super_column.columns[0].name)
        self.assertEqual("Mina", super_column.columns[0].value)

        # add a 'country' subcolumn
        column = Column(name="country", value="Canada", timestamp=100)
        client.batch_mutate(
            {"mina": {"Users": [Mutation(ColumnOrSuperColumn(super_column=SuperColumn("attrs", [column])))]}},
            ThriftConsistencyLevel.ONE,
        )

        super_columns = client.get_slice("mina", column_parent, predicate, ThriftConsistencyLevel.ONE)
        self.assertEqual(1, len(super_columns))
        super_column = super_columns[0].super_column
        self.assertEqual("attrs", super_column.name)
        self.assertEqual(2, len(super_column.columns))

        self.assertEqual("country", super_column.columns[0].name)
        self.assertEqual("Canada", super_column.columns[0].value)

        self.assertEqual("name", super_column.columns[1].name)
        self.assertEqual("Mina", super_column.columns[1].value)

        # add a 'region' subcolumn
        column = Column(name="region", value="Quebec", timestamp=100)
        client.batch_mutate(
            {"mina": {"Users": [Mutation(ColumnOrSuperColumn(super_column=SuperColumn("attrs", [column])))]}},
            ThriftConsistencyLevel.ONE,
        )

        super_columns = client.get_slice("mina", column_parent, predicate, ThriftConsistencyLevel.ONE)
        self.assertEqual(1, len(super_columns))
        super_column = super_columns[0].super_column
        self.assertEqual("attrs", super_column.name)
        self.assertEqual(3, len(super_column.columns))

        self.assertEqual("country", super_column.columns[0].name)
        self.assertEqual("Canada", super_column.columns[0].value)

        self.assertEqual("name", super_column.columns[1].name)
        self.assertEqual("Mina", super_column.columns[1].value)

        self.assertEqual("region", super_column.columns[2].name)
        self.assertEqual("Quebec", super_column.columns[2].value)

    def sc_with_row_cache_test(self):
        """ Test for bug reported in #4190 """
        cluster = self.cluster

        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]
        self.patient_cql_connection(node1)

        node = self.cluster.nodelist()[0]
        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()

        ksdef = KsDef()
        ksdef.name = 'ks'
        ksdef.strategy_class = 'SimpleStrategy'
        ksdef.strategy_options = {'replication_factor': '1'}
        ksdef.cf_defs = []

        client.system_add_keyspace(ksdef)
        client.set_keyspace('ks')

        # create a super column family with UTF8 for all types
        cfdef = CfDef()
        cfdef.keyspace = 'ks'
        cfdef.name = 'Users'
        cfdef.column_type = 'Super'
        cfdef.comparator_type = 'UTF8Type'
        cfdef.subcomparator_type = 'UTF8Type'
        cfdef.key_validation_class = 'UTF8Type'
        cfdef.default_validation_class = 'UTF8Type'
        cfdef.caching = 'rows_only'

        client.system_add_column_family(cfdef)

        column = Column(name='name', value='Mina', timestamp=100)
        client.batch_mutate(
            {'mina': {'Users': [Mutation(ColumnOrSuperColumn(super_column=SuperColumn('attrs', [column])))]}},
            ThriftConsistencyLevel.ONE)

        column_parent = ColumnParent(column_family='Users')
        predicate = SlicePredicate(slice_range=SliceRange("", "", False, 100))
        super_columns = client.get_slice('mina', column_parent, predicate, ThriftConsistencyLevel.ONE)
        self.assertEqual(1, len(super_columns))
        super_column = super_columns[0].super_column
        self.assertEqual('attrs', super_column.name)
        self.assertEqual(1, len(super_column.columns))
        self.assertEqual('name', super_column.columns[0].name)
        self.assertEqual('Mina', super_column.columns[0].value)

        # add a 'country' subcolumn
        column = Column(name='country', value='Canada', timestamp=100)
        client.batch_mutate(
            {'mina': {'Users': [Mutation(ColumnOrSuperColumn(super_column=SuperColumn('attrs', [column])))]}},
            ThriftConsistencyLevel.ONE)

        super_columns = client.get_slice('mina', column_parent, predicate, ThriftConsistencyLevel.ONE)
        self.assertEqual(1, len(super_columns))
        super_column = super_columns[0].super_column
        self.assertEqual('attrs', super_column.name)
        self.assertEqual(2, len(super_column.columns))

        self.assertEqual('country', super_column.columns[0].name)
        self.assertEqual('Canada', super_column.columns[0].value)

        self.assertEqual('name', super_column.columns[1].name)
        self.assertEqual('Mina', super_column.columns[1].value)

        # add a 'region' subcolumn
        column = Column(name='region', value='Quebec', timestamp=100)
        client.batch_mutate(
            {'mina': {'Users': [Mutation(ColumnOrSuperColumn(super_column=SuperColumn('attrs', [column])))]}},
            ThriftConsistencyLevel.ONE)

        super_columns = client.get_slice('mina', column_parent, predicate, ThriftConsistencyLevel.ONE)
        self.assertEqual(1, len(super_columns))
        super_column = super_columns[0].super_column
        self.assertEqual('attrs', super_column.name)
        self.assertEqual(3, len(super_column.columns))

        self.assertEqual('country', super_column.columns[0].name)
        self.assertEqual('Canada', super_column.columns[0].value)

        self.assertEqual('name', super_column.columns[1].name)
        self.assertEqual('Mina', super_column.columns[1].value)

        self.assertEqual('region', super_column.columns[2].name)
        self.assertEqual('Quebec', super_column.columns[2].value)

    def upgrade_with_index_creation_test(self):
        cluster = self.cluster

        # Forcing cluster version on purpose
        cluster.set_install_dir(version="1.2.16")
        if "memtable_allocation_type" in cluster._config_options:
            cluster._config_options.__delitem__("memtable_allocation_type")
        cluster.populate(2).start()

        node1, node2 = cluster.nodelist()

        # wait for the rpc server to start
        session = self.patient_exclusive_cql_connection(node1)

        host, port = node1.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()

        ksdef = KsDef()
        ksdef.name = 'test'
        ksdef.strategy_class = 'SimpleStrategy'
        ksdef.strategy_options = {'replication_factor': '2'}
        ksdef.durable_writes = True
        ksdef.cf_defs = []

        client.system_add_keyspace(ksdef)

        session.cluster.control_connection.wait_for_schema_agreement()

        client.set_keyspace('test')

        # create a super column family with UTF8 for all types
        cfdef = CfDef()
        cfdef.keyspace = 'test'
        cfdef.name = 'sc_test'
        cfdef.column_type = 'Super'
        cfdef.comparator_type = 'UTF8Type'
        cfdef.subcomparator_type = 'UTF8Type'
        cfdef.key_validation_class = 'UTF8Type'
        cfdef.default_validation_class = 'UTF8Type'
        cfdef.caching = 'rows_only'

        client.system_add_column_family(cfdef)

        session.cluster.control_connection.wait_for_schema_agreement()

        for i in range(2):
            supercol_name = 'sc%d' % i
            for j in range(2):
                col_name = 'c%d' % j
                column = Column(name=col_name, value='v', timestamp=100)
                client.batch_mutate(
                    {'k0': {'sc_test': [Mutation(ColumnOrSuperColumn(super_column=SuperColumn(supercol_name, [column])))]}},
                    ThriftConsistencyLevel.ONE)

        session.cluster.shutdown()
        client.transport.close()

        # Upgrade nodes to 2.0 for intermediate sstable conversion
        # See CASSANDRA-7008
        self.upgrade_to_version("binary:2.0.17")
        time.sleep(.5)

        # Upgrade node 1
        node1.flush()
        time.sleep(.5)
        node1.stop(wait_other_notice=True)
        self.set_node_to_current_version(node1)
        node1.start(wait_other_notice=True, wait_for_binary_proto=True)
        time.sleep(.5)

        # wait for the RPC server to start
        session = self.patient_exclusive_cql_connection(node1)

        client = get_thrift_client(host, port)
        client.transport.open()
        client.set_keyspace('test')

        # fetch all supercolumns
        column_parent = ColumnParent(column_family='sc_test')
        predicate = SlicePredicate(slice_range=SliceRange("", "", False, 100))
        super_columns = client.get_slice('k0', column_parent, predicate, ThriftConsistencyLevel.QUORUM)
        self.assertEqual(2, len(super_columns))
        for i in range(2):
            super_column = super_columns[i].super_column
            self.assertEqual('sc%d' % i, super_column.name)
            self.assertEqual(2, len(super_column.columns))
            for j in range(2):
                column = super_column.columns[j]
                self.assertEqual('c%d' % j, column.name)
                self.assertEqual('v', column.value)

        # fetch a single supercolumn
        column_parent = ColumnParent(column_family='sc_test', super_column='sc1')
        columns = client.get_slice('k0', column_parent, predicate, ThriftConsistencyLevel.QUORUM)
        self.assertEqual(2, len(columns))
        for j in range(2):
            column = columns[j].column
            self.assertEqual('c%d' % j, column.name)
            self.assertEqual('v', column.value)
#.........这里部分代码省略.........

    def upgrade_with_counters_test(self):
        cluster = self.cluster

        # Forcing cluster version on purpose
        cluster.set_install_dir(version="1.2.19")
        if "memtable_allocation_type" in cluster._config_options:
            cluster._config_options.__delitem__("memtable_allocation_type")
        cluster.populate(3).start()

        node1, node2, node3 = cluster.nodelist()

        # wait for the rpc server to start
        session = self.patient_exclusive_cql_connection(node1)

        host, port = node1.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()

        ksdef = KsDef()
        ksdef.name = 'test'
        ksdef.strategy_class = 'SimpleStrategy'
        ksdef.strategy_options = {'replication_factor': '2'}
        ksdef.durable_writes = True
        ksdef.cf_defs = []

        client.system_add_keyspace(ksdef)
        client.set_keyspace('test')

        # create a super column family with UTF8 for all types except for the
        # values, which are counters
        cfdef = CfDef()
        cfdef.keyspace = 'test'
        cfdef.name = 'sc_test'
        cfdef.column_type = 'Super'
        cfdef.comparator_type = 'UTF8Type'
        cfdef.subcomparator_type = 'UTF8Type'
        cfdef.key_validation_class = 'UTF8Type'
        cfdef.default_validation_class = 'CounterColumnType'

        client.system_add_column_family(cfdef)

        session.cluster.control_connection.wait_for_schema_agreement()

        for i in range(2):
            supercol_name = 'sc%d' % i
            column_parent = ColumnParent(column_family='sc_test', super_column=supercol_name)
            for j in range(2):
                col_name = 'c%d' % j
                column = CounterColumn(name=col_name, value=1)
                for k in range(20):
                    try:
                        client.add('Counter1', column_parent, column, ThriftConsistencyLevel.ONE)
                    except TimedOutException:
                        raise RerunTestException("re-run test to verify")

        # If we are on 2.1 or any higher version upgrade to 2.0.latest.
        # Otherwise, we must be on a 2.0.x, so we should be upgrading to that version.
        # This will let us test upgrading from 1.2.19 to each of the 2.0 minor releases.
        # Upgrade nodes to 2.0.
        # See CASSANDRA-7008
        self.upgrade_to_version("binary:2.0.17", [node1])
        time.sleep(.5)

        # wait for the RPC server to start
        session = self.patient_exclusive_cql_connection(node1)

        for node in (node1, node2, node3):
            host, port = node.network_interfaces['thrift']
            client = get_thrift_client(host, port)
            client.transport.open()
            client.set_keyspace('test')
            for i in range(2):
                supercol_name = 'sc%d' % i
                column_parent = ColumnParent(column_family='sc_test', super_column=supercol_name)
                for j in range(2):
                    col_name = 'c%d' % j
                    column = CounterColumn(name=col_name, value=1)
                    for k in range(50):
                        try:
                            client.add('Counter1', column_parent, column, ThriftConsistencyLevel.ONE)
                        except TimedOutException:
                            raise RerunTestException("re-run test to verify")

            client.transport.close()

        # Upgrade nodes to 2.0.
        # See CASSANDRA-7008
        self.upgrade_to_version("binary:2.0.17", [node2, node3])
        time.sleep(.5)

        host, port = node1.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()
        client.set_keyspace('test')

        column_parent = ColumnParent(column_family='sc_test')
        predicate = SlicePredicate(slice_range=SliceRange("", "", False, 100))
        super_columns = client.get_slice('Counter1', column_parent, predicate, ThriftConsistencyLevel.QUORUM)
        self.assertEqual(2, len(super_columns))
        for i in range(2):
#.........这里部分代码省略.........

    def upgrade_with_counters_test(self):
        cluster = self.cluster

        # Forcing cluster version on purpose
        cluster.set_install_dir(version="1.2.19")
        cluster.populate(3).start()

        node1, node2, node3 = cluster.nodelist()

        # wait for the rpc server to start
        self.patient_cql_connection(node1)

        host, port = node1.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()

        ksdef = KsDef()
        ksdef.name = 'test'
        ksdef.strategy_class = 'SimpleStrategy'
        ksdef.strategy_options = {'replication_factor': '2'}
        ksdef.durable_writes = True
        ksdef.cf_defs = []

        client.system_add_keyspace(ksdef)
        client.set_keyspace('test')

        # create a super column family with UTF8 for all types except for the
        # values, which are counters
        cfdef = CfDef()
        cfdef.keyspace = 'test'
        cfdef.name = 'sc_test'
        cfdef.column_type = 'Super'
        cfdef.comparator_type = 'UTF8Type'
        cfdef.subcomparator_type = 'UTF8Type'
        cfdef.key_validation_class = 'UTF8Type'
        cfdef.default_validation_class = 'CounterColumnType'

        client.system_add_column_family(cfdef)

        for i in range(2):
            supercol_name = 'sc%d' % i
            column_parent = ColumnParent(column_family='sc_test', super_column=supercol_name)
            for j in range(2):
                col_name = 'c%d' % j
                column = CounterColumn(name=col_name, value=1)
                for k in range(20):
                    client.add('Counter1', column_parent, column, ThriftConsistencyLevel.ONE)

        # If we are on 2.1 or any higher version upgrade to 2.0.latest.
        # Otherwise, we must be on a 2.0.x, so we should be upgrading to that version.
        # This will let us test upgrading from 1.2.19 to each of the 2.0 minor releases.
        CASSANDRA_DIR = os.environ.get('CASSANDRA_DIR')
        if get_version_from_build(CASSANDRA_DIR) >= '2.1':
            # Upgrade nodes to 2.0.
            # See CASSANDRA-7008
            self.upgrade_to_version("git:cassandra-2.0", [node1])
            time.sleep(.5)
        else:
            node1.drain()
            node1.watch_log_for("DRAINED")
            node1.stop(wait_other_notice=False)
            self.set_node_to_current_version(node1)
            node1.start(wait_other_notice=True)

        for node in (node1, node2, node3):
            host, port = node.network_interfaces['thrift']
            client = get_thrift_client(host, port)
            client.transport.open()
            client.set_keyspace('test')
            for i in range(2):
                supercol_name = 'sc%d' % i
                column_parent = ColumnParent(column_family='sc_test', super_column=supercol_name)
                for j in range(2):
                    col_name = 'c%d' % j
                    column = CounterColumn(name=col_name, value=1)
                    for k in range(50):
                        client.add('Counter1', column_parent, column, ThriftConsistencyLevel.ONE)

            client.transport.close()

        if get_version_from_build(CASSANDRA_DIR) >= '2.1':
            # Upgrade nodes to 2.0.
            # See CASSANDRA-7008
            self.upgrade_to_version("git:cassandra-2.0", [node2, node3])
            time.sleep(.5)
        else:
            node2.drain()
            node3.drain()
            node2.watch_log_for("DRAINED")
            node3.watch_log_for("DRAINED")
            node2.stop(wait_other_notice=False)
            node3.stop(wait_other_notice=False)
            self.set_node_to_current_version(node2)
            self.set_node_to_current_version(node3)
            node2.start(wait_other_notice=True)
            node3.start(wait_other_notice=True)

        host, port = node.network_interfaces['thrift']
        client = get_thrift_client(host, port)
        client.transport.open()
#.........这里部分代码省略.........