# coding: utf-8

import itertools
import struct
import time

from cassandra import ConsistencyLevel, InvalidRequest
from cassandra.metadata import NetworkTopologyStrategy, SimpleStrategy
from cassandra.policies import FallthroughRetryPolicy
from cassandra.protocol import ProtocolException
from cassandra.query import SimpleStatement

from dtest import ReusableClusterTester, debug, Tester, create_ks
from distutils.version import LooseVersion
from thrift_bindings.v22.ttypes import \
    ConsistencyLevel as ThriftConsistencyLevel
from thrift_bindings.v22.ttypes import (CfDef, Column, ColumnOrSuperColumn,
                                        Mutation)
from thrift_tests import get_thrift_client
from tools.assertions import (assert_all, assert_invalid, assert_length_equal,
                              assert_none, assert_one, assert_unavailable)
from tools.data import rows_to_list
from tools.decorators import since
from tools.metadata_wrapper import (UpdatingClusterMetadataWrapper,
                                    UpdatingKeyspaceMetadataWrapper,
                                    UpdatingTableMetadataWrapper)


class CQLTester(Tester):

    def prepare(self, ordered=False, create_keyspace=True, use_cache=False,
                nodes=1, rf=1, protocol_version=None, user=None, password=None,
                start_rpc=False, **kwargs):
        cluster = self.cluster

        if ordered:
            cluster.set_partitioner("org.apache.cassandra.dht.ByteOrderedPartitioner")

        if use_cache:
            cluster.set_configuration_options(values={'row_cache_size_in_mb': 100})

        if start_rpc:
            cluster.set_configuration_options(values={'start_rpc': True})

        if user:
            config = {'authenticator': 'org.apache.cassandra.auth.PasswordAuthenticator',
                      'authorizer': 'org.apache.cassandra.auth.CassandraAuthorizer',
                      'permissions_validity_in_ms': 0}
            cluster.set_configuration_options(values=config)

        if not cluster.nodelist():
            cluster.populate(nodes).start(wait_for_binary_proto=True)
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1, protocol_version=protocol_version, user=user, password=password)
        if create_keyspace:
            create_ks(session, 'ks', rf)
        return session


class StorageProxyCQLTester(CQLTester):
    """
    Each CQL statement is exercised at least once in order to
    ensure we execute the code path in StorageProxy.
    # TODO This probably isn't true anymore?
    Note that in depth CQL validation is done in Java unit tests,
    see CASSANDRA-9160.

    # TODO I'm not convinced we need these. Seems like all the functionality
    #      is covered in greater detail in other test classes.
    """

    def keyspace_test(self):
        """
        Smoke test that basic keyspace operations work:

        - create a keyspace
        - assert keyspace exists and is configured as expected with the driver metadata API
        - ALTER it
        - assert keyspace was correctly altered with the driver metadata API
        - DROP it
        - assert keyspace is no longer in keyspace metadata
        """
        session = self.prepare(create_keyspace=False)
        meta = UpdatingClusterMetadataWrapper(session.cluster)

        self.assertNotIn('ks', meta.keyspaces)
        session.execute("CREATE KEYSPACE ks WITH replication = "
                        "{ 'class':'SimpleStrategy', 'replication_factor':1} "
                        "AND DURABLE_WRITES = true")
        self.assertIn('ks', meta.keyspaces)

        ks_meta = UpdatingKeyspaceMetadataWrapper(session.cluster, ks_name='ks')
        self.assertTrue(ks_meta.durable_writes)
        self.assertIsInstance(ks_meta.replication_strategy, SimpleStrategy)

        session.execute("ALTER KEYSPACE ks WITH replication = "
                        "{ 'class' : 'NetworkTopologyStrategy', 'datacenter1' : 1 } "
                        "AND DURABLE_WRITES = false")
        self.assertFalse(ks_meta.durable_writes)
        self.assertIsInstance(ks_meta.replication_strategy, NetworkTopologyStrategy)

        session.execute("DROP KEYSPACE ks")
        self.assertNotIn('ks', meta.keyspaces)

    def table_test(self):
        """
        Smoke test that basic table operations work:

        - create a table
        - ALTER the table adding a column
        - insert 10 values
        - SELECT * and assert the values are there
        - TRUNCATE the table
        - SELECT * and assert there are no values
        - DROP the table
        - SELECT * and assert the statement raises an InvalidRequest
        # TODO run SELECTs to make sure each statement works
        """
        session = self.prepare()

        ks_meta = UpdatingKeyspaceMetadataWrapper(session.cluster, ks_name='ks')

        session.execute("CREATE TABLE test1 (k int PRIMARY KEY, v1 int)")
        self.assertIn('test1', ks_meta.tables)

        t1_meta = UpdatingTableMetadataWrapper(session.cluster, ks_name='ks', table_name='test1')

        session.execute("ALTER TABLE test1 ADD v2 int")
        self.assertIn('v2', t1_meta.columns)

        for i in range(0, 10):
            session.execute("INSERT INTO test1 (k, v1, v2) VALUES ({i}, {i}, {i})".format(i=i))

        assert_all(session, "SELECT * FROM test1", [[i, i, i] for i in range(0, 10)], ignore_order=True)

        session.execute("TRUNCATE test1")

        assert_none(session, "SELECT * FROM test1")

        session.execute("DROP TABLE test1")
        self.assertNotIn('test1', ks_meta.tables)

    @since("2.0", max_version="3.X")
    def table_test_compact_storage(self):
        """
        Smoke test that basic table operations work:

        - create a table with COMPACT STORAGE
        - insert 10 values
        - SELECT * and assert the values are there
        - TRUNCATE the table
        - SELECT * and assert there are no values
        - DROP the table
        - SELECT * and assert the statement raises an InvalidRequest
        # TODO run SELECTs to make sure each statement works
        """
        session = self.prepare()

        ks_meta = UpdatingKeyspaceMetadataWrapper(session.cluster, ks_name='ks')

        session.execute("CREATE TABLE test2 (k int, c1 int, v1 int, PRIMARY KEY (k, c1)) WITH COMPACT STORAGE")
        self.assertIn('test2', ks_meta.tables)

        for i in range(0, 10):
            session.execute("INSERT INTO test2 (k, c1, v1) VALUES ({i}, {i}, {i})".format(i=i))

        assert_all(session, "SELECT * FROM test2", [[i, i, i] for i in range(0, 10)], ignore_order=True)

        session.execute("TRUNCATE test2")

        assert_none(session, "SELECT * FROM test2")

        session.execute("DROP TABLE test2")
        self.assertNotIn('test2', ks_meta.tables)

    def index_test(self):
        """
        Smoke test CQL statements related to indexes:

        - CREATE a table
        - CREATE an index on that table
        - INSERT 10 values into the table
        - SELECT from the table over the indexed value and assert the expected values come back
        - drop the index
        - assert SELECTing over the indexed value raises an InvalidRequest
        # TODO run SELECTs to make sure each statement works
        """
        session = self.prepare()

        session.execute("CREATE TABLE test3 (k int PRIMARY KEY, v1 int, v2 int)")
        table_meta = UpdatingTableMetadataWrapper(session.cluster, ks_name='ks', table_name='test3')
        session.execute("CREATE INDEX testidx ON test3 (v1)")
        self.assertIn('testidx', table_meta.indexes)

        for i in range(0, 10):
            session.execute("INSERT INTO test3 (k, v1, v2) VALUES ({i}, {i}, {i})".format(i=i))

        assert_one(session, "SELECT * FROM test3 WHERE v1 = 0", [0, 0, 0])

        session.execute("DROP INDEX testidx")
        self.assertNotIn('testidx', table_meta.indexes)

    def type_test(self):
        """
        Smoke test basic TYPE operations:

        - CREATE a type
        - CREATE a table using that type
        - ALTER the type and CREATE another table
        - DROP the tables and type
        - CREATE another table using the DROPped type and assert it fails with an InvalidRequest
        # TODO run SELECTs to make sure each statement works
        # TODO is this even necessary given the existence of the auth_tests?
        """
        session = self.prepare()
        # even though we only ever use the user_types attribute of this object,
        # we have to access it each time, because attribute access is how the
        # value is updated
        ks_meta = UpdatingKeyspaceMetadataWrapper(session.cluster, ks_name='ks')

        session.execute("CREATE TYPE address_t (street text, city text, zip_code int)")
        self.assertIn('address_t', ks_meta.user_types)

        session.execute("CREATE TABLE test4 (id int PRIMARY KEY, address frozen<address_t>)")

        session.execute("ALTER TYPE address_t ADD phones set<text>")
        self.assertIn('phones', ks_meta.user_types['address_t'].field_names)

        # drop the table so we can safely drop the type it uses
        session.execute("DROP TABLE test4")

        session.execute("DROP TYPE address_t")
        self.assertNotIn('address_t', ks_meta.user_types)

    def user_test(self):
        """
        Smoke test for basic USER queries:

        - get a session as the default superuser
        - CREATE a user
        - ALTER that user by giving it a different password
        - DROP that user
        # TODO list users after each to make sure each statement works
        """
        session = self.prepare(user='cassandra', password='cassandra')
        node1 = self.cluster.nodelist()[0]

        def get_usernames():
            return [user.name for user in session.execute('LIST USERS')]

        self.assertNotIn('user1', get_usernames())

        session.execute("CREATE USER user1 WITH PASSWORD 'secret'")
        # use patient to retry until it works, because it takes some time for
        # the CREATE to take
        self.patient_cql_connection(node1, user='user1', password='secret')

        session.execute("ALTER USER user1 WITH PASSWORD 'secret^2'")
        # use patient for same reason as above
        self.patient_cql_connection(node1, user='user1', password='secret^2')

        session.execute("DROP USER user1")
        self.assertNotIn('user1', get_usernames())

    def statements_test(self):
        """
        Smoke test SELECT and UPDATE statements:

        - create a table
        - insert 20 rows into the table
        - run SELECT COUNT queries and assert they return the correct values
            - bare and with IN and equality conditions
        - run SELECT * queries with = conditions
        - run UPDATE queries
        - SELECT * and assert the UPDATEd values are there
        - DELETE with a = condition
        - SELECT the deleted values and make sure nothing is returned
        # TODO run SELECTs to make sure each statement works
        """
        session = self.prepare()

        session.execute("CREATE TABLE test7 (kind text, time int, v1 int, v2 int, PRIMARY KEY(kind, time) )")

        for i in range(0, 10):
            session.execute("INSERT INTO test7 (kind, time, v1, v2) VALUES ('ev1', {i}, {i}, {i})".format(i=i))
            session.execute("INSERT INTO test7 (kind, time, v1, v2) VALUES ('ev2', {i}, {i}, {i})".format(i=i))

        assert_one(session, "SELECT COUNT(*) FROM test7 WHERE kind = 'ev1'", [10])

        assert_one(session, "SELECT COUNT(*) FROM test7 WHERE kind IN ('ev1', 'ev2')", [20])

        assert_one(session, "SELECT COUNT(*) FROM test7 WHERE kind IN ('ev1', 'ev2') AND time=0", [2])

        assert_all(session, "SELECT * FROM test7 WHERE kind = 'ev1'", [['ev1', i, i, i] for i in range(0, 10)])

        assert_all(session, "SELECT * FROM test7 WHERE kind = 'ev2'", [['ev2', i, i, i] for i in range(0, 10)])

        for i in range(0, 10):
            session.execute("UPDATE test7 SET v1 = 0, v2 = 0 where kind = 'ev1' AND time={i}".format(i=i))

        assert_all(session, "SELECT * FROM test7 WHERE kind = 'ev1'", [['ev1', i, 0, 0] for i in range(0, 10)])

        session.execute("DELETE FROM test7 WHERE kind = 'ev1'")
        assert_none(session, "SELECT * FROM test7 WHERE kind = 'ev1'")

        assert_one(session, "SELECT COUNT(*) FROM test7 WHERE kind = 'ev1'", [0])

    @since('3.10')
    def partition_key_allow_filtering_test(self):
        """
        Filtering with unrestricted parts of partition keys

        @jira_ticket CASSANDRA-11031
        """
        session = self.prepare()

        session.execute("""
            CREATE TABLE IF NOT EXISTS test_filter (
                k1 int,
                k2 int,
                ck1 int,
                v int,
                PRIMARY KEY ((k1, k2), ck1)
            )
        """)

        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 0, 0, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 0, 1, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 0, 2, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 0, 3, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 1, 0, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 1, 1, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 1, 2, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (0, 1, 3, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 0, 0, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 0, 1, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 0, 2, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 0, 3, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 1, 0, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 1, 1, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 1, 2, 0)")
        session.execute("INSERT INTO test_filter (k1, k2, ck1, v) VALUES (1, 1, 3, 0)")

        # select test
        assert_all(session,
                   "SELECT * FROM test_filter WHERE k1 = 0 ALLOW FILTERING",
                   [[0, 0, 0, 0],
                    [0, 0, 1, 0],
                    [0, 0, 2, 0],
                    [0, 0, 3, 0],
                    [0, 1, 0, 0],
                    [0, 1, 1, 0],
                    [0, 1, 2, 0],
                    [0, 1, 3, 0]],
                   ignore_order=True)

        assert_all(session,
                   "SELECT * FROM test_filter WHERE k1 <= 1 AND k2 >= 1 ALLOW FILTERING",
                   [[0, 1, 0, 0],
                    [0, 1, 1, 0],
                    [0, 1, 2, 0],
                    [0, 1, 3, 0],
                    [1, 1, 0, 0],
                    [1, 1, 1, 0],
                    [1, 1, 2, 0],
                    [1, 1, 3, 0]],
                   ignore_order=True)

        assert_none(session, "SELECT * FROM test_filter WHERE k1 = 2 ALLOW FILTERING")
        assert_none(session, "SELECT * FROM test_filter WHERE k1 <=0 AND k2 > 1 ALLOW FILTERING")

        assert_all(session,
                   "SELECT * FROM test_filter WHERE k2 <= 0 ALLOW FILTERING",
                   [[0, 0, 0, 0],
                    [0, 0, 1, 0],
                    [0, 0, 2, 0],
                    [0, 0, 3, 0],
                    [1, 0, 0, 0],
                    [1, 0, 1, 0],
                    [1, 0, 2, 0],
                    [1, 0, 3, 0]],
                   ignore_order=True)

        assert_all(session,
                   "SELECT * FROM test_filter WHERE k1 <= 0 AND k2 = 0 ALLOW FILTERING",
                   [[0, 0, 0, 0],
                    [0, 0, 1, 0],
                    [0, 0, 2, 0],
                    [0, 0, 3, 0]])

        assert_all(session,
                   "SELECT * FROM test_filter WHERE k2 = 1 ALLOW FILTERING",
                   [[0, 1, 0, 0],
                    [0, 1, 1, 0],
                    [0, 1, 2, 0],
                    [0, 1, 3, 0],
                    [1, 1, 0, 0],
                    [1, 1, 1, 0],
                    [1, 1, 2, 0],
                    [1, 1, 3, 0]],
                   ignore_order=True)

        assert_none(session, "SELECT * FROM test_filter WHERE k2 = 2 ALLOW FILTERING")

        # filtering on both Partition Key and Clustering key
        assert_all(session,
                   "SELECT * FROM test_filter WHERE k1 = 0 AND ck1=0 ALLOW FILTERING",
                   [[0, 0, 0, 0],
                    [0, 1, 0, 0]],
                   ignore_order=True)

        assert_all(session,
                   "SELECT * FROM test_filter WHERE k1 = 0 AND k2=1 AND ck1=0 ALLOW FILTERING",
                   [[0, 1, 0, 0]])

        # count(*) test
        assert_all(session,
                   "SELECT count(*) FROM test_filter WHERE k2 = 0 ALLOW FILTERING",
                   [[8]])

        assert_all(session,
                   "SELECT count(*) FROM test_filter WHERE k2 = 1 ALLOW FILTERING",
                   [[8]])

        assert_all(session,
                   "SELECT count(*) FROM test_filter WHERE k2 = 2 ALLOW FILTERING",
                   [[0]])

        # test invalid query
        with self.assertRaises(InvalidRequest):
            session.execute("SELECT * FROM test_filter WHERE k1 = 0")

        with self.assertRaises(InvalidRequest):
            session.execute("SELECT * FROM test_filter WHERE k1 = 0 AND k2 > 0")

        with self.assertRaises(InvalidRequest):
            session.execute("SELECT * FROM test_filter WHERE k1 >= 0 AND k2 in (0,1,2)")

        with self.assertRaises(InvalidRequest):
            session.execute("SELECT * FROM test_filter WHERE k2 > 0")

    def batch_test(self):
        """
        Smoke test for BATCH statements:

        - CREATE a table
        - create a BATCH statement and execute it at QUORUM
        # TODO run SELECTs to make sure each statement works
        """
        session = self.prepare()

        session.execute("""
            CREATE TABLE test8 (
                userid text PRIMARY KEY,
                name text,
                password text
            )
        """)

        query = SimpleStatement("""
            BEGIN BATCH
                INSERT INTO test8 (userid, password, name) VALUES ('user2', 'ch@ngem3b', 'second user');
                UPDATE test8 SET password = 'ps22dhds' WHERE userid = 'user3';
                INSERT INTO test8 (userid, password) VALUES ('user4', 'ch@ngem3c');
                DELETE name FROM test8 WHERE userid = 'user1';
            APPLY BATCH;
        """, consistency_level=ConsistencyLevel.QUORUM)
        session.execute(query)


class MiscellaneousCQLTester(CQLTester):
    """
    CQL tests that cannot be performed as Java unit tests, see CASSANDRA-9160.
    If you're considering adding a test here, consider writing Java unit tests
    for CQL validation instead. Add a new test here only if there is a reason
    for it, e.g. the test is related to the client protocol or thrift, requires
    examining the log files, or must run on multiple nodes.
    """

    @since('2.1', max_version='3.0')
    def large_collection_errors_test(self):
        """
        Assert C* logs warnings when selecting too large a collection over
        protocol v2:

        - prepare the cluster and connect using protocol v2
        - CREATE a table containing a map column
        - insert over 65535 elements into the map
        - select all the elements of the map
        - assert that the correct error was logged
        """

        # We only warn with protocol 2
        session = self.prepare(protocol_version=2)

        cluster = self.cluster
        node1 = cluster.nodelist()[0]
        self.ignore_log_patterns = ["Detected collection for table"]

        session.execute("""
            CREATE TABLE maps (
                userid text PRIMARY KEY,
                properties map<int, text>
            );
        """)

        # Insert more than the max, which is 65535
        for i in range(70000):
            session.execute("UPDATE maps SET properties[{}] = 'x' WHERE userid = 'user'".format(i))

        # Query for the data and throw exception
        session.execute("SELECT properties FROM maps WHERE userid = 'user'")
        node1.watch_log_for("Detected collection for table ks.maps with 70000 elements, more than the 65535 limit. "
                            "Only the first 65535 elements will be returned to the client. Please see "
                            "http://cassandra.apache.org/doc/cql3/CQL.html#collections for more details.")

    @since('2.0', max_version='4')
    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])

    @since('2.0', max_version='4')
    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 invalid_string_literals_test(self):
        """
        @jira_ticket CASSANDRA-8101

        - assert INSERTing into a nonexistent table fails normally, with an InvalidRequest exception
        - create a table with ascii and text columns
        - assert that trying to execute an insert statement with non-UTF8 contents raises a ProtocolException
            - tries to insert into a nonexistent column to make sure the ProtocolException is raised over other errors
        """
        session = self.prepare()
        # this should fail as normal, not with a ProtocolException
        assert_invalid(session, u"insert into invalid_string_literals (k, a) VALUES (0, '\u038E\u0394\u03B4\u03E0')")

        session = self.patient_cql_connection(self.cluster.nodelist()[0], keyspace='ks')
        session.execute("create table invalid_string_literals (k int primary key, a ascii, b text)")

        # this should still fail with an InvalidRequest
        assert_invalid(session, u"insert into invalid_string_literals (k, c) VALUES (0, '\u038E\u0394\u03B4\u03E0')")
        # but since the protocol requires strings to be valid UTF-8, the error
        # response to this is a ProtocolException, not an error about the
        # nonexistent column
        with self.assertRaisesRegexp(ProtocolException, 'Cannot decode string as UTF8'):
            session.execute("insert into invalid_string_literals (k, c) VALUES (0, '\xc2\x01')")

    def prepared_statement_invalidation_test(self):
        """
        @jira_ticket CASSANDRA-7910

        - CREATE a table and INSERT a row
        - prepare 2 prepared SELECT statements
        - SELECT the row with a bound prepared statement and assert it returns the expected row
        - ALTER the table, dropping a column
        - assert prepared statement without that column in it still works
        - assert prepared statement containing that column fails
        - ALTER the table, adding a column
        - assert prepared statement without that column in it still works
        - assert prepared statement containing that column also still works
        - ALTER the table, changing the type of a column
        - assert that both prepared statements still work
        """
        session = self.prepare()

        session.execute("CREATE TABLE test (k int PRIMARY KEY, a int, b int, c int)")
        session.execute("INSERT INTO test (k, a, b, c) VALUES (0, 0, 0, 0)")

        wildcard_prepared = session.prepare("SELECT * FROM test")
        explicit_prepared = session.prepare("SELECT k, a, b, c FROM test")
        result = session.execute(wildcard_prepared.bind(None))
        self.assertEqual(result, [(0, 0, 0, 0)])

        session.execute("ALTER TABLE test DROP c")
        result = session.execute(wildcard_prepared.bind(None))
        # wildcard select can be automatically re-prepared by the driver
        self.assertEqual(result, [(0, 0, 0)])
        # but re-preparing the statement with explicit columns should fail
        # (see PYTHON-207 for why we expect InvalidRequestException instead of the normal exc)
        assert_invalid(session, explicit_prepared.bind(None), expected=InvalidRequest)

        session.execute("ALTER TABLE test ADD d int")
        result = session.execute(wildcard_prepared.bind(None))
        self.assertEqual(result, [(0, 0, 0, None)])

        if self.cluster.version() < LooseVersion('3.0'):
            explicit_prepared = session.prepare("SELECT k, a, b, d FROM test")

            # when the type is altered, both statements will need to be re-prepared
            # by the driver, but the re-preparation should succeed
            session.execute("ALTER TABLE test ALTER d TYPE blob")
            result = session.execute(wildcard_prepared.bind(None))
            self.assertEqual(result, [(0, 0, 0, None)])

            result = session.execute(explicit_prepared.bind(None))
            self.assertEqual(result, [(0, 0, 0, None)])

    def range_slice_test(self):
        """
        Regression test for CASSANDRA-1337:

        - CREATE a table
        - INSERT 2 rows
        - SELECT * from the table
        - assert 2 rows were returned

        @jira_ticket CASSANDRA-1337
        # TODO I don't see how this is an interesting test or how it tests 1337.
        """

        cluster = self.cluster

        cluster.populate(2).start()
        node1 = cluster.nodelist()[0]
        time.sleep(0.2)

        session = self.patient_cql_connection(node1)
        create_ks(session, 'ks', 1)

        session.execute("""
            CREATE TABLE test (
                k text PRIMARY KEY,
                v int
            );
        """)
        time.sleep(1)

        session.execute("INSERT INTO test (k, v) VALUES ('foo', 0)")
        session.execute("INSERT INTO test (k, v) VALUES ('bar', 1)")

        res = list(session.execute("SELECT * FROM test"))
        self.assertEqual(len(res), 2, msg=res)

    def many_columns_test(self):
        """
        Test for tables with thousands of columns.
        For CASSANDRA-11621.
        """

        session = self.prepare()
        width = 5000
        cluster = self.cluster

        session.execute("CREATE TABLE very_wide_table (pk int PRIMARY KEY, " +
                        ",".join(map(lambda i: "c_{} int".format(i), range(width))) +
                        ")")

        session.execute("INSERT INTO very_wide_table (pk, " +
                        ",".join(map(lambda i: "c_{}".format(i), range(width))) +
                        ") VALUES (100," +
                        ",".join(map(lambda i: str(i), range(width))) +
                        ")")

        assert_all(session, "SELECT " +
                   ",".join(map(lambda i: "c_{}".format(i), range(width))) +
                   " FROM very_wide_table", [[i for i in range(width)]])

    @since("3.11", max_version="3.X")
    def drop_compact_storage_flag_test(self):
        """
        Test for CASSANDRA-10857, verifying the schema change
        distribution across the other nodes.

        """

        cluster = self.cluster

        cluster.populate(3).start()
        node1, node2, node3 = cluster.nodelist()

        session1 = self.patient_cql_connection(node1)
        session2 = self.patient_cql_connection(node2)
        session3 = self.patient_cql_connection(node3)
        create_ks(session1, 'ks', 3)
        sessions = [session1, session2, session3]

        for session in sessions:
            session.set_keyspace('ks')

        session1.execute("""
            CREATE TABLE test_drop_compact_storage (k int PRIMARY KEY, s1 int) WITH COMPACT STORAGE;
        """)

        session1.execute("INSERT INTO test_drop_compact_storage (k, s1) VALUES (1,1)")
        session1.execute("INSERT INTO test_drop_compact_storage (k, s1) VALUES (2,2)")
        session1.execute("INSERT INTO test_drop_compact_storage (k, s1) VALUES (3,3)")

        for session in sessions:
            res = session.execute("SELECT * from test_drop_compact_storage")
            self.assertEqual(rows_to_list(res), [[1, 1],
                                                 [2, 2],
                                                 [3, 3]])

        session1.execute("ALTER TABLE test_drop_compact_storage DROP COMPACT STORAGE")

        for session in sessions:
            assert_all(session, "SELECT * from test_drop_compact_storage",
                       [[1, None, 1, None],
                        [2, None, 2, None],
                        [3, None, 3, None]])


@since('3.2')
class AbortedQueryTester(CQLTester):
    """
    @jira_ticket CASSANDRA-7392

    Test that read-queries that take longer than read_request_timeout_in_ms
    time out.

    # TODO The important part of these is "set up a combination of
    #      configuration options that will make all reads time out, then
    #      try to read and assert it times out". This can probably be made much
    #      simpler -- most of the logic can be factored out. In many cases it
    #      probably isn't even necessary to define a custom table or to insert
    #      more than one value.
    """

    def local_query_test(self):
        """
        Check that a query running on the local coordinator node times out:

        - set the read request timeouts to 1 second
        - start the cluster with read_iteration_delay set to 5 ms
            - the delay will be applied ot each row iterated and will cause
              read queries to take longer than the read timeout
        - CREATE and INSERT into a table
        - SELECT * from the table using a retry policy that never retries, and assert it times out

        @jira_ticket CASSANDRA-7392
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'request_timeout_in_ms': 1000,
                                                  'read_request_timeout_in_ms': 1000,
                                                  'range_request_timeout_in_ms': 1000})

        # cassandra.test.read_iteration_delay_ms causes the state tracking read iterators
        # introduced by CASSANDRA-7392 to pause by the specified amount of milliseconds every
        # CQL row iterated for non system queries, so that these queries take much longer to complete,
        # see ReadCommand.withStateTracking()
        cluster.populate(1).start(wait_for_binary_proto=True,
                                  jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                                            "-Dcassandra.test.read_iteration_delay_ms=5"])
        node = cluster.nodelist()[0]
        session = self.patient_cql_connection(node)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test1 (
                id int PRIMARY KEY,
                val text
            );
        """)

        for i in range(500):
            session.execute("INSERT INTO test1 (id, val) VALUES ({}, 'foo')".format(i))

        # use debug logs because at info level no-spam logger has unpredictable results
        mark = node.mark_log(filename='debug.log')
        statement = SimpleStatement("SELECT * from test1",
                                    consistency_level=ConsistencyLevel.ONE,
                                    retry_policy=FallthroughRetryPolicy())
        assert_unavailable(lambda c: debug(c.execute(statement)), session)
        node.watch_log_for("operations timed out", filename='debug.log', from_mark=mark, timeout=60)

    def remote_query_test(self):
        """
        Check that a query running on a node other than the coordinator times out:

        - populate the cluster with 2 nodes
        - set the read request timeouts to 1 second
        - start one node without having it join the ring
        - start the other node with read_iteration_delay set to 5 ms
            - the delay will be applied ot each row iterated and will cause
              read queries to take longer than the read timeout
        - CREATE a table
        - INSERT 5000 rows on a session on the node that is not a member of the ring
        - run SELECT statements and assert they fail
        # TODO refactor SELECT statements:
        #        - run the statements in a loop to reduce duplication
        #        - watch the log after each query
        #        - assert we raise the right error
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'request_timeout_in_ms': 1000,
                                                  'read_request_timeout_in_ms': 1000,
                                                  'range_request_timeout_in_ms': 1000})

        cluster.populate(2)
        node1, node2 = cluster.nodelist()

        node1.start(wait_for_binary_proto=True, join_ring=False)  # ensure other node executes queries
        node2.start(wait_for_binary_proto=True,
                    jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                              "-Dcassandra.test.read_iteration_delay_ms=5"])  # see above for explanation

        session = self.patient_exclusive_cql_connection(node1)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test2 (
                id int,
                col int,
                val text,
                PRIMARY KEY(id, col)
            );
        """)

        for i, j in itertools.product(range(10), range(500)):
            session.execute("INSERT INTO test2 (id, col, val) VALUES ({}, {}, 'foo')".format(i, j))

        # use debug logs because at info level no-spam logger has unpredictable results
        mark = node2.mark_log(filename='debug.log')

        statement = SimpleStatement("SELECT * from test2",
                                    consistency_level=ConsistencyLevel.ONE,
                                    retry_policy=FallthroughRetryPolicy())
        assert_unavailable(lambda c: debug(c.execute(statement)), session)

        statement = SimpleStatement("SELECT * from test2 where id = 1",
                                    consistency_level=ConsistencyLevel.ONE,
                                    retry_policy=FallthroughRetryPolicy())
        assert_unavailable(lambda c: debug(c.execute(statement)), session)

        statement = SimpleStatement("SELECT * from test2 where id IN (1, 2, 3) AND col > 10",
                                    consistency_level=ConsistencyLevel.ONE,
                                    retry_policy=FallthroughRetryPolicy())
        assert_unavailable(lambda c: debug(c.execute(statement)), session)

        statement = SimpleStatement("SELECT * from test2 where col > 5 ALLOW FILTERING",
                                    consistency_level=ConsistencyLevel.ONE,
                                    retry_policy=FallthroughRetryPolicy())
        assert_unavailable(lambda c: debug(c.execute(statement)), session)

        node2.watch_log_for("operations timed out", filename='debug.log', from_mark=mark, timeout=60)

    def index_query_test(self):
        """
        Check that a secondary index query times out:

        - populate a 1-node cluster
        - set the read request timeouts to 1 second
        - start one node without having it join the ring
        - start the other node with read_iteration_delay set to 5 ms
            - the delay will be applied ot each row iterated and will cause
              read queries to take longer than the read timeout
        - CREATE a table
        - CREATE an index on the table
        - INSERT 500 values into the table
        - SELECT over the table and assert it times out
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'request_timeout_in_ms': 1000,
                                                  'read_request_timeout_in_ms': 1000,
                                                  'range_request_timeout_in_ms': 1000})

        cluster.populate(1).start(wait_for_binary_proto=True,
                                  jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                                            "-Dcassandra.test.read_iteration_delay_ms=5"])  # see above for explanation
        node = cluster.nodelist()[0]
        session = self.patient_cql_connection(node)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test3 (
                id int PRIMARY KEY,
                col int,
                val text
            );
        """)

        session.execute("CREATE INDEX ON test3 (col)")

        for i in range(500):
            session.execute("INSERT INTO test3 (id, col, val) VALUES ({}, 50, 'foo')".format(i))

        # use debug logs because at info level no-spam logger has unpredictable results
        mark = node.mark_log(filename='debug.log')
        statement = session.prepare("SELECT * from test3 WHERE col = ? ALLOW FILTERING")
        statement.consistency_level = ConsistencyLevel.ONE
        statement.retry_policy = FallthroughRetryPolicy()
        assert_unavailable(lambda c: debug(c.execute(statement, [50])), session)
        node.watch_log_for("operations timed out", filename='debug.log', from_mark=mark, timeout=60)

    def materialized_view_test(self):
        """
        Check that a materialized view query times out:

        - populate a 2-node cluster
        - set the read request timeouts to 1 second
        - start one node without having it join the ring
        - start the other node with read_iteration_delay set to 5 ms
            - the delay will be applied ot each row iterated and will cause
              read queries to take longer than the read timeout
        - CREATE a table
        - INSERT 500 values into that table
        - CREATE a materialized view over that table
        - assert querying that table results in an unavailable exception
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'request_timeout_in_ms': 1000,
                                                  'read_request_timeout_in_ms': 1000,
                                                  'range_request_timeout_in_ms': 1000})

        cluster.populate(2)
        node1, node2 = cluster.nodelist()

        node1.start(wait_for_binary_proto=True, join_ring=False)  # ensure other node executes queries
        node2.start(wait_for_binary_proto=True,
                    jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                              "-Dcassandra.test.read_iteration_delay_ms=5"])  # see above for explanation

        session = self.patient_exclusive_cql_connection(node1)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test4 (
                id int PRIMARY KEY,
                col int,
                val text
            );
        """)

        session.execute(("CREATE MATERIALIZED VIEW mv AS SELECT * FROM test4 "
                         "WHERE col IS NOT NULL AND id IS NOT NULL PRIMARY KEY (col, id)"))

        for i in range(500):
            session.execute("INSERT INTO test4 (id, col, val) VALUES ({}, 50, 'foo')".format(i))

        # use debug logs because at info level no-spam logger has unpredictable results
        mark = node2.mark_log(filename='debug.log')
        statement = SimpleStatement("SELECT * FROM mv WHERE col = 50",
                                    consistency_level=ConsistencyLevel.ONE,
                                    retry_policy=FallthroughRetryPolicy())

        assert_unavailable(lambda c: debug(c.execute(statement)), session)
        node2.watch_log_for("operations timed out", filename='debug.log', from_mark=mark, timeout=60)


@since('3.10')
class SlowQueryTester(CQLTester):
    """
    Test slow query logging.

    @jira_ticket CASSANDRA-12403
    """
    def local_query_test(self):
        """
        Check that a query running locally on the coordinator is reported as slow:

        - start a one node cluster with slow_query_log_timeout_in_ms set to a small value
          and the read request timeouts set to a large value (to ensure the query is not aborted) and
          read_iteration_delay set to a value big enough for the query to exceed slow_query_log_timeout_in_ms
          (this will cause read queries to take longer than the slow query timeout)
        - CREATE and INSERT into a table
        - SELECT * from the table using a retry policy that never retries, and check that the slow
          query log messages are present in the debug logs (we cannot check the logs at info level because
          the no spam logger has unpredictable results)

        @jira_ticket CASSANDRA-12403
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'slow_query_log_timeout_in_ms': 10,
                                                  'request_timeout_in_ms': 120000,
                                                  'read_request_timeout_in_ms': 120000,
                                                  'range_request_timeout_in_ms': 120000})

        # cassandra.test.read_iteration_delay_ms causes the state tracking read iterators
        # introduced by CASSANDRA-7392 to pause by the specified amount of milliseconds during each
        # iteration of non system queries, so that these queries take much longer to complete,
        # see ReadCommand.withStateTracking()
        cluster.populate(1).start(wait_for_binary_proto=True,
                                  jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                                            "-Dcassandra.test.read_iteration_delay_ms=1"])
        node = cluster.nodelist()[0]
        session = self.patient_cql_connection(node)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test1 (
                id int,
                col int,
                val text,
                PRIMARY KEY(id, col)
            );
        """)

        for i in range(100):
            session.execute("INSERT INTO test1 (id, col, val) VALUES (1, {}, 'foo')".format(i))

        # only check debug logs because at INFO level the no-spam logger has unpredictable results
        mark = node.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test1",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node.watch_log_for(["operations were slow", "SELECT \* FROM ks.test1"],
                           from_mark=mark, filename='debug.log', timeout=60)
        mark = node.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test1 where id = 1",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node.watch_log_for(["operations were slow", "SELECT \* FROM ks.test1"],
                           from_mark=mark, filename='debug.log', timeout=60)
        mark = node.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test1 where id = 1",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node.watch_log_for(["operations were slow", "SELECT \* FROM ks.test1"],
                           from_mark=mark, filename='debug.log', timeout=60)
        mark = node.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test1 where token(id) < 0",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node.watch_log_for(["operations were slow", "SELECT \* FROM ks.test1"],
                           from_mark=mark, filename='debug.log', timeout=60)

    def remote_query_test(self):
        """
        Check that a query running on a node other than the coordinator is reported as slow:

        - populate the cluster with 2 nodes
        - start one node without having it join the ring
        - start the other one node with slow_query_log_timeout_in_ms set to a small value
          and the read request timeouts set to a large value (to ensure the query is not aborted) and
          read_iteration_delay set to a value big enough for the query to exceed slow_query_log_timeout_in_ms
          (this will cause read queries to take longer than the slow query timeout)
        - CREATE a table
        - INSERT 5000 rows on a session on the node that is not a member of the ring
        - run SELECT statements and check that the slow query messages are present in the debug logs
          (we cannot check the logs at info level because the no spam logger has unpredictable results)

        @jira_ticket CASSANDRA-12403
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'slow_query_log_timeout_in_ms': 10,
                                                  'request_timeout_in_ms': 120000,
                                                  'read_request_timeout_in_ms': 120000,
                                                  'range_request_timeout_in_ms': 120000})

        cluster.populate(2)
        node1, node2 = cluster.nodelist()

        node1.start(wait_for_binary_proto=True, join_ring=False)  # ensure other node executes queries
        node2.start(wait_for_binary_proto=True,
                    jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                              "-Dcassandra.test.read_iteration_delay_ms=1"])  # see above for explanation

        session = self.patient_exclusive_cql_connection(node1)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test2 (
                id int,
                col int,
                val text,
                PRIMARY KEY(id, col)
            );
        """)

        for i, j in itertools.product(range(100), range(10)):
            session.execute("INSERT INTO test2 (id, col, val) VALUES ({}, {}, 'foo')".format(i, j))

        # only check debug logs because at INFO level the no-spam logger has unpredictable results
        mark = node2.mark_log(filename='debug.log')
        session.execute(SimpleStatement("SELECT * from test2",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node2.watch_log_for(["operations were slow", "SELECT \* FROM ks.test2"],
                            from_mark=mark, filename='debug.log', timeout=60)
        mark = node2.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test2 where id = 1",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node2.watch_log_for(["operations were slow", "SELECT \* FROM ks.test2"],
                            from_mark=mark, filename='debug.log', timeout=60)
        mark = node2.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test2 where id = 1",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node2.watch_log_for(["operations were slow", "SELECT \* FROM ks.test2"],
                            from_mark=mark, filename='debug.log', timeout=60)
        mark = node2.mark_log(filename='debug.log')

        session.execute(SimpleStatement("SELECT * from test2 where token(id) < 0",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        node2.watch_log_for(["operations were slow", "SELECT \* FROM ks.test2"],
                            from_mark=mark, filename='debug.log', timeout=60)

    def disable_slow_query_log_test(self):
        """
        Check that a query is NOT reported as slow if slow query logging is disabled.

        - start a one node cluster with slow_query_log_timeout_in_ms set to 0 milliseconds
          (this will disable slow query logging), the read request timeouts set to a large value
          (to ensure queries are not aborted) and read_iteration_delay set to 5 milliseconds
          (this will cause read queries to take longer than usual)
        - CREATE and INSERT into a table
        - SELECT * from the table using a retry policy that never retries, and check that the slow
          query log messages are present in the logs

        @jira_ticket CASSANDRA-12403
        """
        cluster = self.cluster
        cluster.set_configuration_options(values={'slow_query_log_timeout_in_ms': 0,
                                                  'request_timeout_in_ms': 120000,
                                                  'read_request_timeout_in_ms': 120000,
                                                  'range_request_timeout_in_ms': 120000})

        # cassandra.test.read_iteration_delay_ms causes the state tracking read iterators
        # introduced by CASSANDRA-7392 to pause by the specified amount of milliseconds during each
        # iteration of non system queries, so that these queries take much longer to complete,
        # see ReadCommand.withStateTracking()
        cluster.populate(1).start(wait_for_binary_proto=True,
                                  jvm_args=["-Dcassandra.monitoring_report_interval_ms=10",
                                            "-Dcassandra.test.read_iteration_delay_ms=1"])
        node = cluster.nodelist()[0]
        session = self.patient_cql_connection(node)

        create_ks(session, 'ks', 1)
        session.execute("""
            CREATE TABLE test3 (
                id int PRIMARY KEY,
                val text
            );
        """)

        for i in range(100):
            session.execute("INSERT INTO test3 (id, val) VALUES ({}, 'foo')".format(i))

        session.execute(SimpleStatement("SELECT * from test3",
                                        consistency_level=ConsistencyLevel.ONE,
                                        retry_policy=FallthroughRetryPolicy()))

        time.sleep(1)  # do our best to ensure logs had a chance to appear

        self._check_logs(node, "SELECT \* FROM ks.test3", 'debug.log', 0)

    def _check_logs(self, node, pattern, filename, num_expected):
        ret = node.grep_log(pattern, filename=filename)
        assert_length_equal(ret, num_expected)


class LWTTester(ReusableClusterTester):
    """
    Validate CQL queries for LWTs for static columns for null and non-existing rows
    @jira_ticket CASSANDRA-9842
    """

    @classmethod
    def post_initialize_cluster(cls):
        cluster = cls.cluster
        cluster.populate(3)
        cluster.start(wait_for_binary_proto=True)

    def get_lwttester_session(self):
        node1 = self.cluster.nodelist()[0]
        session = self.patient_cql_connection(node1)
        session.execute("""CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION={'class':'SimpleStrategy',
            'replication_factor':1}""")
        session.execute("USE ks")
        return session

    def lwt_with_static_columns_test(self):
        session = self.get_lwttester_session()

        session.execute("""
            CREATE TABLE lwt_with_static (a int, b int, s int static, d text, PRIMARY KEY (a, b))
        """)

        assert_one(session, "UPDATE lwt_with_static SET s = 1 WHERE a = 1 IF s = NULL", [True])

        assert_one(session, "SELECT * FROM lwt_with_static", [1, None, 1, None])

        assert_one(session, "UPDATE lwt_with_static SET s = 2 WHERE a = 2 IF EXISTS", [False])

        assert_one(session, "SELECT * FROM lwt_with_static WHERE a = 1", [1, None, 1, None])

        assert_one(session, "INSERT INTO lwt_with_static (a, s) VALUES (2, 2) IF NOT EXISTS", [True])

        assert_one(session, "SELECT * FROM lwt_with_static WHERE a = 2", [2, None, 2, None])

        assert_one(session, "BEGIN BATCH\n" +
                   "INSERT INTO lwt_with_static (a, b, d) values (3, 3, 'a');\n" +
                   "UPDATE lwt_with_static SET s = 3 WHERE a = 3 IF s = null;\n" +
                   "APPLY BATCH;", [True])

        assert_one(session, "SELECT * FROM lwt_with_static WHERE a = 3", [3, 3, 3, "a"])

        # LWT applies before INSERT
        assert_one(session, "BEGIN BATCH\n" +
                   "INSERT INTO lwt_with_static (a, b, d) values (4, 4, 'a');\n" +
                   "UPDATE lwt_with_static SET s = 4 WHERE a = 4 IF s = null;\n" +
                   "APPLY BATCH;", [True])

        assert_one(session, "SELECT * FROM lwt_with_static WHERE a = 4", [4, 4, 4, "a"])

    def _validate_non_existing_or_null_values(self, table_name, session):
        assert_one(session, "UPDATE {} SET s = 1 WHERE a = 1 IF s = NULL".format(table_name), [True])

        assert_one(session, "SELECT a, s, d FROM {} WHERE a = 1".format(table_name), [1, 1, None])

        assert_one(session, "UPDATE {} SET s = 2 WHERE a = 2 IF s IN (10,20,NULL)".format(table_name), [True])

        assert_one(session, "SELECT a, s, d FROM {} WHERE a = 2".format(table_name), [2, 2, None])

        assert_one(session, "UPDATE {} SET s = 4 WHERE a = 4 IF s != 4".format(table_name), [True])

        assert_one(session, "SELECT a, s, d FROM {} WHERE a = 4".format(table_name), [4, 4, None])

    def _is_new_lwt_format_version(self, version):
        return version > LooseVersion('3.9') or (version > LooseVersion('3.0.9') and version < LooseVersion('3.1'))

    def conditional_updates_on_static_columns_with_null_values_test(self):
        session = self.get_lwttester_session()

        table_name = "conditional_updates_on_static_columns_with_null"
        session.execute("""
            CREATE TABLE {} (a int, b int, s int static, d text, PRIMARY KEY (a, b))
        """.format(table_name))

        for i in range(1, 6):
            session.execute("INSERT INTO {} (a, b) VALUES ({}, {})".format(table_name, i, i))

        self._validate_non_existing_or_null_values(table_name, session)

        assert_one(session, "UPDATE {} SET s = 30 WHERE a = 3 IF s IN (10,20,30)".format(table_name),
                   [False, None] if self._is_new_lwt_format_version(self.cluster.version()) else [False])

        assert_one(session, "SELECT * FROM {} WHERE a = 3".format(table_name), [3, 3, None, None])

        for operator in [">", "<", ">=", "<=", "="]:
            assert_one(session, "UPDATE {} SET s = 50 WHERE a = 5 IF s {} 3".format(table_name, operator),
                       [False, None] if self._is_new_lwt_format_version(self.cluster.version()) else [False])

            assert_one(session, "SELECT * FROM {} WHERE a = 5".format(table_name), [5, 5, None, None])

    def conditional_updates_on_static_columns_with_non_existing_values_test(self):
        session = self.get_lwttester_session()

        table_name = "conditional_updates_on_static_columns_with_ne"
        session.execute("""
            CREATE TABLE {} (a int, b int, s int static, d text, PRIMARY KEY (a, b))
        """.format(table_name))

        self._validate_non_existing_or_null_values(table_name, session)

        assert_one(session, "UPDATE {} SET s = 30 WHERE a = 3 IF s IN (10,20,30)".format(table_name), [False])

        assert_none(session, "SELECT * FROM {} WHERE a = 3".format(table_name))

        for operator in [">", "<", ">=", "<=", "="]:
            assert_one(session, "UPDATE {} SET s = 50 WHERE a = 5 IF s {} 3".format(table_name, operator), [False])

            assert_none(session, "SELECT * FROM {} WHERE a = 5".format(table_name))

    def _validate_non_existing_or_null_values_batch(self, table_name, session):
        assert_one(session, """
            BEGIN BATCH
                INSERT INTO {table_name} (a, b, d) values (2, 2, 'a');
                UPDATE {table_name} SET s = 2 WHERE a = 2 IF s = null;
            APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {table_name} WHERE a = 2".format(table_name=table_name), [2, 2, 2, "a"])

        assert_one(session, """
            BEGIN BATCH
                INSERT INTO {table_name} (a, b, s, d) values (4, 4, 4, 'a')
                UPDATE {table_name} SET s = 5 WHERE a = 4 IF s = null;
            APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {table_name} WHERE a = 4".format(table_name=table_name), [4, 4, 5, "a"])

        assert_one(session, """
            BEGIN BATCH
                INSERT INTO {table_name} (a, b, s, d) values (5, 5, 5, 'a')
                UPDATE {table_name} SET s = 6 WHERE a = 5 IF s IN (1,2,null)
            APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {table_name} WHERE a = 5".format(table_name=table_name), [5, 5, 6, "a"])

        assert_one(session, """
            BEGIN BATCH
                INSERT INTO {table_name} (a, b, s, d) values (7, 7, 7, 'a')
                UPDATE {table_name} SET s = 8 WHERE a = 7 IF s != 7;
            APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {table_name} WHERE a = 7".format(table_name=table_name), [7, 7, 8, "a"])

    def conditional_updates_on_static_columns_with_null_values_batch_test(self):
        session = self.get_lwttester_session()

        table_name = "lwt_on_static_columns_with_null_batch"
        session.execute("""
            CREATE TABLE {table_name} (a int, b int, s int static, d text, PRIMARY KEY (a, b))
        """.format(table_name=table_name))

        for i in range(1, 7):
            session.execute("INSERT INTO {table_name} (a, b) VALUES ({i}, {i})".format(table_name=table_name, i=i))

        self._validate_non_existing_or_null_values_batch(table_name, session)

        for operator in [">", "<", ">=", "<=", "="]:
            assert_one(session, """
                BEGIN BATCH
                    INSERT INTO {table_name} (a, b, s, d) values (3, 3, 40, 'a')
                    UPDATE {table_name} SET s = 30 WHERE a = 3 IF s {operator} 5;
                APPLY BATCH""".format(table_name=table_name, operator=operator),
                       [False, 3, 3, None] if self._is_new_lwt_format_version(self.cluster.version()) else [False])

            assert_one(session, "SELECT * FROM {table_name} WHERE a = 3".format(table_name=table_name), [3, 3, None, None])

        assert_one(session, """
                BEGIN BATCH
                    INSERT INTO {table_name} (a, b, s, d) values (6, 6, 70, 'a')
                    UPDATE {table_name} SET s = 60 WHERE a = 6 IF s IN (1,2,3)
                APPLY BATCH""".format(table_name=table_name),
                   [False, 6, 6, None] if self._is_new_lwt_format_version(self.cluster.version()) else [False])

        assert_one(session, "SELECT * FROM {table_name} WHERE a = 6".format(table_name=table_name), [6, 6, None, None])

    def conditional_deletes_on_static_columns_with_null_values_test(self):
        session = self.get_lwttester_session()

        table_name = "conditional_deletes_on_static_with_null"
        session.execute("""
            CREATE TABLE {} (a int, b int, s1 int static, s2 int static, v int, PRIMARY KEY (a, b))
        """.format(table_name))

        for i in range(1, 6):
            session.execute("INSERT INTO {} (a, b, s1, s2, v) VALUES ({}, {}, {}, null, {})".format(table_name, i, i, i, i))

        assert_one(session, "DELETE s1 FROM {} WHERE a = 1 IF s2 = null".format(table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 1".format(table_name), [1, 1, None, None, 1])

        assert_one(session, "DELETE s1 FROM {} WHERE a = 2 IF s2 IN (10,20,30)".format(table_name), [False, None])

        assert_one(session, "SELECT * FROM {} WHERE a = 2".format(table_name), [2, 2, 2, None, 2])

        assert_one(session, "DELETE s1 FROM {} WHERE a = 3 IF s2 IN (null,20,30)".format(table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 3".format(table_name), [3, 3, None, None, 3])

        assert_one(session, "DELETE s1 FROM {} WHERE a = 4 IF s2 != 4".format(table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 4".format(table_name), [4, 4, None, None, 4])

        for operator in [">", "<", ">=", "<=", "="]:
            assert_one(session, "DELETE s1 FROM {} WHERE a = 5 IF s2 {} 3".format(table_name, operator), [False, None])
            assert_one(session, "SELECT * FROM {} WHERE a = 5".format(table_name), [5, 5, 5, None, 5])

    def conditional_deletes_on_static_columns_with_null_values_batch_test(self):
        session = self.get_lwttester_session()

        table_name = "conditional_deletes_on_static_with_null_batch"
        session.execute("""
            CREATE TABLE {} (a int, b int, s1 int static, s2 int static, v int, PRIMARY KEY (a, b))
        """.format(table_name))

        assert_one(session, """
             BEGIN BATCH
                 INSERT INTO {table_name} (a, b, s1, v) values (2, 2, 2, 2);
                 DELETE s1 FROM {table_name} WHERE a = 2 IF s2 = null;
             APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 2".format(table_name), [2, 2, None, None, 2])

        for operator in [">", "<", ">=", "<=", "="]:
            assert_one(session, """
                BEGIN BATCH
                    INSERT INTO {table_name} (a, b, s1, v) values (3, 3, 3, 3);
                    DELETE s1 FROM {table_name} WHERE a = 3 IF s2 {operator} 5;
                APPLY BATCH""".format(table_name=table_name, operator=operator), [False])

            assert_none(session, "SELECT * FROM {} WHERE a = 3".format(table_name))

        assert_one(session, """
             BEGIN BATCH
                 INSERT INTO {table_name} (a, b, s1, v) values (6, 6, 6, 6);
                 DELETE s1 FROM {table_name} WHERE a = 6 IF s2 IN (1,2,3);
             APPLY BATCH""".format(table_name=table_name), [False])

        assert_none(session, "SELECT * FROM {} WHERE a = 6".format(table_name))

        assert_one(session, """
             BEGIN BATCH
                 INSERT INTO {table_name} (a, b, s1, v) values (4, 4, 4, 4);
                 DELETE s1 FROM {table_name} WHERE a = 4 IF s2 = null;
             APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 4".format(table_name), [4, 4, None, None, 4])

        assert_one(session, """
            BEGIN BATCH
                INSERT INTO {table_name} (a, b, s1, v) VALUES (5, 5, 5, 5);
                DELETE s1 FROM {table_name} WHERE a = 5 IF s1 IN (1,2,null);
            APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 5".format(table_name), [5, 5, None, None, 5])

        assert_one(session, """
            BEGIN BATCH
                INSERT INTO {table_name} (a, b, s1, v) values (7, 7, 7, 7);
                DELETE s1 FROM {table_name} WHERE a = 7 IF s2 != 7;
            APPLY BATCH""".format(table_name=table_name), [True])

        assert_one(session, "SELECT * FROM {} WHERE a = 7".format(table_name), [7, 7, None, None, 7])

    def lwt_with_empty_resultset(self):
        """
        LWT with unset row.
        @jira_ticket CASSANDRA-12694
        """
        session = self.get_lwttester_session()

        session.execute("""
            CREATE TABLE test (pk text, v1 int, v2 text, PRIMARY KEY (pk));
        """)
        session.execute("update test set v1 = 100 where pk = 'test1';")
        node1 = self.cluster.nodelist()[0]
        self.cluster.flush()
        assert_one(session, "UPDATE test SET v1 = 100 WHERE pk = 'test1' IF v2 = null;", [True])