text: import time
  import uuid
  import pytest
  import logging
  
  from flaky import flaky
  
  from cassandra import ConsistencyLevel, Unauthorized, InvalidRequest
  from cassandra.query import SimpleStatement
  
  from dtest import Tester, create_ks
  from tools.assertions import assert_invalid
  from tools.flaky import retry
  from plugins.assert_tools import assert_regexp_matches
  
  since = pytest.mark.since
  logger = logging.getLogger(__name__)
  
  
  def listify(item):
      """
      listify a query result consisting of user types
  
      returns nested arrays representing user type ordering
      """
      if isinstance(item, (tuple, list)):
          return [listify(i) for i in item]
      else:
          return item
  
  
  class TestUserTypes(Tester):
      def assertUnauthorized(self, session, query, message):
          with pytest.raises(Unauthorized) as cm:
              session.execute(query)
              assert_regexp_matches(repr(cm._excinfo[1]), message)
  
      def assertNoTypes(self, session):
          for keyspace in list(session.cluster.metadata.keyspaces.values()):
              assert 0 == len(keyspace.user_types)
  
      def test_type_dropping(self):
          """
          Tests that a type cannot be dropped when in use, and otherwise can be dropped.
          """
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_type_dropping', 2)
  
          stmt = """
                USE user_type_dropping
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TYPE simple_type (
                user_number int
                )
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TABLE simple_table (
                id uuid PRIMARY KEY,
                number frozen<simple_type>
                )
             """
          session.execute(stmt)
          # Make sure the schema propagate
          time.sleep(2)
  
          _id = uuid.uuid4()
          stmt = """
                INSERT INTO simple_table (id, number)
                VALUES ({id}, {{user_number: 1}});
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                DROP TYPE simple_type;
             """
          assert_invalid(session, stmt, 'Cannot drop user type user_type_dropping.simple_type as it is still used by table user_type_dropping.simple_table')
  
          # now that we've confirmed that a user type cannot be dropped while in use
          # let's remove the offending table
  
          # TODO: uncomment below after CASSANDRA-6472 is resolved
          # and add another check to make sure the table/type drops succeed
          stmt = """
                DROP TABLE simple_table;
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                DROP TYPE simple_type;
             """
          session.execute(stmt)
  
          # now let's have a look at the system schema and make sure no user types are defined
          self.assertNoTypes(session)
  
      def test_nested_type_dropping(self):
          """
          Confirm a user type can't be dropped when being used by another user type.
          """
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'nested_user_type_dropping', 2)
  
          stmt = """
                USE nested_user_type_dropping
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TYPE simple_type (
                user_number int,
                user_text text
                )
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TYPE another_type (
                somefield frozen<simple_type>
                )
             """
          session.execute(stmt)
  
          stmt = """
                DROP TYPE simple_type;
             """
          assert_invalid(session, stmt, 'Cannot drop user type nested_user_type_dropping.simple_type as it is still used by user type another_type')
  
          # drop the type that's impeding the drop, and then try again
          stmt = """
                DROP TYPE another_type;
             """
          session.execute(stmt)
  
          stmt = """
                DROP TYPE simple_type;
             """
          session.execute(stmt)
  
          # now let's have a look at the system schema and make sure no user types are defined
          self.assertNoTypes(session)
  
      def test_type_enforcement(self):
          """
          Confirm error when incorrect data type used for user type
          """
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_type_enforcement', 2)
  
          stmt = """
                USE user_type_enforcement
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TYPE simple_type (
                user_number int
                )
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TABLE simple_table (
                id uuid PRIMARY KEY,
                number frozen<simple_type>
                )
             """
          session.execute(stmt)
          # Make sure the schema propagate
          time.sleep(2)
  
          # here we will attempt an insert statement which should fail
          # because the user type is an int, but the insert statement is
          # providing text
          _id = uuid.uuid4()
          stmt = """
                INSERT INTO simple_table (id, number)
                VALUES ({id}, {{user_number: 'uh oh....this is not a number'}});
             """.format(id=_id)
          assert_invalid(session, stmt, 'field user_number is not of type int')
  
          # let's check the rowcount and make sure the data
          # didn't get inserted when the exception asserted above was thrown
          stmt = """
                SELECT * FROM simple_table;
             """
          rows = list(session.execute(stmt))
          assert 0 == len(rows)
  
      def test_nested_user_types(self):
          """Tests user types within user types"""
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_types', 2)
  
          stmt = """
                USE user_types
             """
          session.execute(stmt)
  
          # Create a user type to go inside another one:
          stmt = """
                CREATE TYPE item (
                sub_one text,
                sub_two text,
                )
             """
          session.execute(stmt)
  
          # Create a user type to contain the item:
          stmt = """
                CREATE TYPE container (
                stuff text,
                more_stuff frozen<item>
                )
             """
          session.execute(stmt)
  
          # Create a table that holds and item, a container, and a
          # list of containers:
          stmt = """
                CREATE TABLE bucket (
                 id uuid PRIMARY KEY,
                 primary_item frozen<item>,
                 other_items frozen<container>,
                 other_containers list<frozen<container>>
                )
             """
          session.execute(stmt)
          # Make sure the schema propagate
          time.sleep(2)
  
          # Insert some data:
          _id = uuid.uuid4()
          stmt = """
                INSERT INTO bucket (id, primary_item)
                VALUES ({id}, {{sub_one: 'test', sub_two: 'test2'}});
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                UPDATE bucket
                SET other_items = {{stuff: 'stuff', more_stuff: {{sub_one: 'one', sub_two: 'two'}}}}
                WHERE id={id};
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                UPDATE bucket
                SET other_containers = other_containers + [
                     {{
                         stuff: 'stuff2',
                         more_stuff: {{sub_one: 'one_other', sub_two: 'two_other'}}
                     }}
                ]
                WHERE id={id};
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                UPDATE bucket
                SET other_containers = other_containers + [
                    {{
                        stuff: 'stuff3',
                        more_stuff: {{sub_one: 'one_2_other', sub_two: 'two_2_other'}}
                    }},
                    {{stuff: 'stuff4',
                      more_stuff: {{sub_one: 'one_3_other', sub_two: 'two_3_other'}}
                    }}
                ]
                WHERE id={id};
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                SELECT primary_item, other_items, other_containers from bucket where id={id};
             """.format(id=_id)
          rows = list(session.execute(stmt))
  
          primary_item, other_items, other_containers = rows[0]
          assert listify(primary_item) == ['test', 'test2']
          assert listify(other_items) == ['stuff', ['one', 'two']]
          assert listify(other_containers) == [['stuff2', ['one_other', 'two_other']], ['stuff3', ['one_2_other', 'two_2_other']], ['stuff4', ['one_3_other', 'two_3_other']]]
  
          #  Generate some repetitive data and check it for it's contents:
          for x in range(50):
  
              # Create row:
              _id = uuid.uuid4()
              stmt = """
                UPDATE bucket
                SET other_containers = other_containers + [
                    {{
                        stuff: 'stuff3',
                        more_stuff: {{
                            sub_one: 'one_2_other', sub_two: 'two_2_other'
                        }}
                    }},
                    {{
                        stuff: 'stuff4',
                        more_stuff: {{
                            sub_one: 'one_3_other', sub_two: 'two_3_other'
                        }}
                    }}
                ]
                WHERE id={id};
             """.format(id=_id)
              session.execute(stmt)
  
              time.sleep(0.1)
  
              # Check it:
              stmt = """
                SELECT other_containers from bucket WHERE id={id}
              """.format(id=_id)
              rows = list(session.execute(stmt))
  
              items = rows[0][0]
              assert listify(items) == [['stuff3', ['one_2_other', 'two_2_other']], ['stuff4', ['one_3_other', 'two_3_other']]]
  
      def test_type_as_part_of_pkey(self):
          """Tests user types as part of a composite pkey"""
          # make sure we can define a table with a user type as part of the pkey
          # and do a basic insert/query of data in that table.
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_type_pkeys', 2)
  
          stmt = """
                CREATE TYPE t_person_name (
                first text,
                middle text,
                last text
              )
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TABLE person_likes (
                id uuid,
                name frozen<t_person_name>,
                like text,
                PRIMARY KEY ((id, name))
                )
             """
          session.execute(stmt)
          # Make sure the schema propagate
          time.sleep(2)
  
          _id = uuid.uuid4()
  
          stmt = """
                INSERT INTO person_likes (id, name, like)
                VALUES ({id}, {{first:'Nero', middle:'Claudius Caesar Augustus', last:'Germanicus'}}, 'arson');
             """.format(id=_id)
          session.execute(stmt)
  
          # attempt to query without the user type portion of the pkey and confirm there is an error
          stmt = """
                SELECT id, name.first from person_likes where id={id};
             """.format(id=_id)
  
          if self.cluster.version() >= '3.10':
              assert_invalid(session, stmt, 'Cannot execute this query as it might involve data filtering')
          elif self.cluster.version() >= '2.2':
              assert_invalid(session, stmt, 'Partition key parts: name must be restricted as other parts are')
          else:
              assert_invalid(session, stmt, 'Partition key part name must be restricted since preceding part is')
  
          stmt = """
                SELECT id, name.first, like from person_likes where id={id} and name = {{first:'Nero', middle: 'Claudius Caesar Augustus', last: 'Germanicus'}};
             """.format(id=_id)
          rows = session.execute(stmt)
  
          row_uuid, first_name, like = rows[0]
          assert first_name == 'Nero'
          assert like == 'arson'
  
      @flaky
      def test_type_secondary_indexing(self):
          """
          Confirm that user types are secondary-indexable
          Similar procedure to TestSecondaryIndexesOnCollections.test_list_indexes
          """
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_type_indexing', 2)
  
          stmt = """
                CREATE TYPE t_person_name (
                first text,
                middle text,
                last text
              )
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TABLE person_likes (
                id uuid PRIMARY KEY,
                name frozen<t_person_name>,
                like text
                )
             """
          session.execute(stmt)
          # Make sure the schema propagate
          time.sleep(2)
  
          # add index and query (even though there are no rows in the table yet)
          stmt = """
                CREATE INDEX person_likes_name on person_likes (name);
              """
          session.execute(stmt)
  
          stmt = """
                SELECT * from person_likes where name = {first:'Nero', middle: 'Claudius Caesar Augustus', last: 'Germanicus'};
              """
          rows = list(session.execute(stmt))
          assert 0 == len(rows)
  
          # add a row which doesn't specify data for the indexed column, and query again
          _id = uuid.uuid4()
          stmt = """
                INSERT INTO person_likes (id, like)
                VALUES ({id}, 'long walks on the beach');
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                SELECT * from person_likes where name = {first:'Bob', middle: 'Testy', last: 'McTesterson'};
              """
  
          rows = list(session.execute(stmt))
          assert 0 == len(rows)
  
          # finally let's add a queryable row, and get it back using the index
          _id = uuid.uuid4()
  
          stmt = """
                INSERT INTO person_likes (id, name, like)
                VALUES ({id}, {{first:'Nero', middle:'Claudius Caesar Augustus', last:'Germanicus'}}, 'arson');
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
                SELECT id, name.first, like from person_likes where name = {first:'Nero', middle: 'Claudius Caesar Augustus', last: 'Germanicus'};
             """
  
          rows = list(session.execute(stmt))
  
          row_uuid, first_name, like = rows[0]
  
          assert str(row_uuid) == str(_id)
          assert first_name == 'Nero'
          assert like == 'arson'
  
          # rename a field in the type and make sure the index still works
          stmt = """
              ALTER TYPE t_person_name rename first to first_name;
              """
          session.execute(stmt)
  
          stmt = """
              SELECT id, name.first_name, like from person_likes where name = {first_name:'Nero', middle: 'Claudius Caesar Augustus', last: 'Germanicus'};
              """
  
          rows = list(session.execute(stmt))
  
          row_uuid, first_name, like = rows[0]
  
          assert str(row_uuid) == str(_id)
          assert first_name == 'Nero'
          assert like == 'arson'
  
          # add another row to be sure the index is still adding new data
          _id = uuid.uuid4()
  
          stmt = """
                INSERT INTO person_likes (id, name, like)
                VALUES ({id}, {{first_name:'Abraham', middle:'', last:'Lincoln'}}, 'preserving unions');
             """.format(id=_id)
          session.execute(stmt)
  
          stmt = """
              SELECT id, name.first_name, like from person_likes where name = {first_name:'Abraham', middle:'', last:'Lincoln'};
              """
  
          rows = list(session.execute(stmt))
  
          row_uuid, first_name, like = rows[0]
  
          assert str(row_uuid) == str(_id)
          assert first_name == 'Abraham'
          assert like == 'preserving unions'
  
      def test_type_keyspace_permission_isolation(self):
          """
          Confirm permissions are respected for types in different keyspaces
          """
          self.fixture_dtest_setup.ignore_log_patterns = [
              # I think this happens when permissions change and a node becomes temporarily unavailable
              # and it's probably ok to ignore on this test, as I can see the schema changes propogating
              # almost immediately after
              r'Can\'t send migration request: node.*is down',
          ]
  
          cluster = self.cluster
          config = {'authenticator': 'org.apache.cassandra.auth.PasswordAuthenticator',
                    'authorizer': 'org.apache.cassandra.auth.CassandraAuthorizer',
                    'permissions_validity_in_ms': 0}
          cluster.set_configuration_options(values=config)
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
  
          # do setup that requires a super user
          superuser_session = self.patient_cql_connection(node1, user='cassandra', password='cassandra')
  
          # role setup is async and does not log when complete; the only way to know its done (on a single instance) is to get a successful return.
          def allowed_error(error):
              return isinstance(error, InvalidRequest) and "Cannot process role related query as the role manager isn't yet setup" in str(error)
          def r(fn):
              retry(fn, allowed_error=allowed_error, sleep_seconds=10)
  
          # role setup  process is local to each instance, so can't rely on the first query passing to say this is complete; node1 may be complete but not node2 and node3
          # for this reason, need to be aggressive about retries
          r(lambda: superuser_session.execute("create user ks1_user with password 'cassandra' nosuperuser;"))
          r(lambda: superuser_session.execute("create user ks2_user with password 'cassandra' nosuperuser;"))
          r(lambda: create_ks(superuser_session, 'ks1', 2))
          r(lambda: create_ks(superuser_session, 'ks2', 2))
          r(lambda: superuser_session.execute("grant all permissions on keyspace ks1 to ks1_user;"))
          r(lambda: superuser_session.execute("grant all permissions on keyspace ks2 to ks2_user;"))
  
          user1_session = self.patient_cql_connection(node1, user='ks1_user', password='cassandra')
          user2_session = self.patient_cql_connection(node1, user='ks2_user', password='cassandra')
  
          # first make sure the users can't create types in each other's ks
          self.assertUnauthorized(user1_session, "CREATE TYPE ks2.simple_type (user_number int, user_text text );", 'User ks1_user has no CREATE permission on <keyspace ks2> or any of its parents')
  
          self.assertUnauthorized(user2_session, "CREATE TYPE ks1.simple_type (user_number int, user_text text );", 'User ks2_user has no CREATE permission on <keyspace ks1> or any of its parents')
  
          # now, actually create the types in the correct keyspaces
          user1_session.execute("CREATE TYPE ks1.simple_type (user_number int, user_text text );")
          user2_session.execute("CREATE TYPE ks2.simple_type (user_number int, user_text text );")
  
          # each user now has a type belonging to their granted keyspace
          # let's make sure they can't drop each other's types (for which they have no permissions)
  
          self.assertUnauthorized(user1_session, "DROP TYPE ks2.simple_type;", 'User ks1_user has no DROP permission on <keyspace ks2> or any of its parents')
  
          self.assertUnauthorized(user2_session, "DROP TYPE ks1.simple_type;", 'User ks2_user has no DROP permission on <keyspace ks1> or any of its parents')
  
          # let's make sure they can't rename each other's types (for which they have no permissions)
          self.assertUnauthorized(user1_session, "ALTER TYPE ks2.simple_type RENAME user_number TO user_num;", 'User ks1_user has no ALTER permission on <keyspace ks2> or any of its parents')
  
          self.assertUnauthorized(user2_session, "ALTER TYPE ks1.simple_type RENAME user_number TO user_num;", 'User ks2_user has no ALTER permission on <keyspace ks1> or any of its parents')
  
          # rename the types using the correct user w/permissions to do so
          user1_session.execute("ALTER TYPE ks1.simple_type RENAME user_number TO user_num;")
          user2_session.execute("ALTER TYPE ks2.simple_type RENAME user_number TO user_num;")
  
          # finally, drop the types using the correct user w/permissions to do so, consistency all avoids using a sleep
          user1_session.execute(SimpleStatement("DROP TYPE ks1.simple_type;", consistency_level=ConsistencyLevel.ALL))
          user2_session.execute(SimpleStatement("DROP TYPE ks2.simple_type;", consistency_level=ConsistencyLevel.ALL))
  
          time.sleep(5)
  
          # verify user type metadata is gone from the system schema
          self.assertNoTypes(superuser_session)
  
      def test_nulls_in_user_types(self):
          """Tests user types with null values"""
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_types', 2)
  
          stmt = """
                USE user_types
             """
          session.execute(stmt)
  
          # Create a user type to go inside another one:
          stmt = """
                CREATE TYPE item (
                sub_one text,
                sub_two text,
                )
             """
          session.execute(stmt)
  
          # Create a table that holds an item
          stmt = """
                CREATE TABLE bucket (
                 id int PRIMARY KEY,
                 my_item frozen<item>,
                )
             """
          session.execute(stmt)
          # Make sure the schema propagates
          time.sleep(2)
  
          # Adds an explicit null
          session.execute("INSERT INTO bucket (id, my_item) VALUES (0, {sub_one: 'test', sub_two: null})")
          # Adds with an implicit null
          session.execute("INSERT INTO bucket (id, my_item) VALUES (1, {sub_one: 'test'})")
  
          rows = list(session.execute("SELECT my_item FROM bucket WHERE id=0"))
          assert listify(rows[0]) == [['test', None]]
  
          rows = list(session.execute("SELECT my_item FROM bucket WHERE id=1"))
          assert listify(rows[0]) == [['test', None]]
  
      def test_no_counters_in_user_types(self):
          # CASSANDRA-7672
          cluster = self.cluster
          cluster.populate(1).start()
          [node1] = cluster.nodelist()
          session = self.patient_cql_connection(node1)
          create_ks(session, 'user_types', 1)
  
          stmt = """
              USE user_types
           """
          session.execute(stmt)
  
          stmt = """
              CREATE TYPE t_item (
              sub_one COUNTER )
           """
  
          assert_invalid(session, stmt, 'A user type cannot contain counters')
  
      def test_type_as_clustering_col(self):
          """Tests user types as clustering column"""
          # make sure we can define a table with a user type as a clustering column
          # and do a basic insert/query of data in that table.
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_type_pkeys', 2)
  
          stmt = """
                CREATE TYPE t_letterpair (
                first text,
                second text
              )
             """
          session.execute(stmt)
  
          stmt = """
                CREATE TABLE letters (
                id int,
                letterpair frozen<t_letterpair>,
                PRIMARY KEY (id, letterpair)
                )
             """
          session.execute(stmt)
          # address CASSANDRA-11498
          session.cluster.control_connection.wait_for_schema_agreement()
  
          # create a bit of data and expect a natural order based on clustering user types
  
          ids = list(range(1, 10))
  
          for _id in ids:
              session.execute("INSERT INTO letters (id, letterpair) VALUES ({}, {{first:'a', second:'z'}})".format(_id))
              session.execute("INSERT INTO letters (id, letterpair) VALUES ({}, {{first:'z', second:'a'}})".format(_id))
              session.execute("INSERT INTO letters (id, letterpair) VALUES ({}, {{first:'c', second:'f'}})".format(_id))
              session.execute("INSERT INTO letters (id, letterpair) VALUES ({}, {{first:'c', second:'a'}})".format(_id))
              session.execute("INSERT INTO letters (id, letterpair) VALUES ({}, {{first:'c', second:'z'}})".format(_id))
              session.execute("INSERT INTO letters (id, letterpair) VALUES ({}, {{first:'d', second:'e'}})".format(_id))
  
          for _id in ids:
              res = list(session.execute("SELECT letterpair FROM letters where id = {}".format(_id)))
  
              assert listify(res) == [[['a', 'z']], [['c', 'a']], [['c', 'f']], [['c', 'z']], [['d', 'e']], [['z', 'a']]]
  
      @since('3.6')
      def test_udt_subfield(self):
          """
          @jira_ticket CASSANDRA-7423
          @since 3.6
          """
          cluster = self.cluster
          cluster.populate(3).start()
          node1, node2, node3 = cluster.nodelist()
          session = self.patient_cql_connection(node1, consistency_level=ConsistencyLevel.LOCAL_QUORUM)
          create_ks(session, 'user_types', 1)
  
          # Check we can create non-frozen table
          session.execute("CREATE TYPE udt (first ascii, second int, third int)")
          session.execute("CREATE TABLE t (id int PRIMARY KEY, v udt)")
  
          # Fill in a full UDT across two statements
          # Ensure all subfields are set
          session.execute("INSERT INTO t (id, v) VALUES (0, {third: 2, second: 1})")
          session.execute("UPDATE t set v.first = 'a' WHERE id=0")
          rows = list(session.execute("SELECT * FROM t WHERE id = 0"))
          assert listify(rows) == [[0, ['a', 1, 2]]]
  
          # Create a full udt
          # Update a subfield on the udt
          # Read back the updated udt
          session.execute("INSERT INTO t (id, v) VALUES (0, {first: 'c', second: 3, third: 33})")
          session.execute("UPDATE t set v.second = 5 where id=0")
          rows = list(session.execute("SELECT * FROM t WHERE id=0"))
          assert listify(rows) == [[0, ['c', 5, 33]]]
  
          # Rewrite the entire udt
          # Read back
          session.execute("INSERT INTO t (id, v) VALUES (0, {first: 'alpha', second: 111, third: 100})")
          rows = list(session.execute("SELECT * FROM t WHERE id=0"))
          assert listify(rows) == [[0, ['alpha', 111, 100]]]
  
          # Send three subfield updates to udt
          # Read back
          session.execute("UPDATE t set v.first = 'beta' WHERE id=0")
          session.execute("UPDATE t set v.first = 'delta' WHERE id=0")
          session.execute("UPDATE t set v.second = -10 WHERE id=0")
          rows = list(session.execute("SELECT * FROM t WHERE id=0"))
          assert listify(rows) == [[0, ['delta', -10, 100]]]
  
          # Send conflicting updates serially to different nodes
          # Read back
          session1 = self.exclusive_cql_connection(node1)
          session2 = self.exclusive_cql_connection(node2)
          session3 = self.exclusive_cql_connection(node3)
  
          session1.execute("UPDATE user_types.t set v.third = 101 WHERE id=0")
          session2.execute("UPDATE user_types.t set v.third = 102 WHERE id=0")
          session3.execute("UPDATE user_types.t set v.third = 103 WHERE id=0")
          query = SimpleStatement("SELECT * FROM t WHERE id = 0", consistency_level=ConsistencyLevel.ALL)
          rows = list(session.execute(query))
          assert listify(rows) == [[0, ['delta', -10, 103]]]
          session1.shutdown()
          session2.shutdown()
          session3.shutdown()
  
          # Write full UDT, set one field to null, read back
          session.execute("INSERT INTO t (id, v) VALUES (0, {first:'cass', second:3, third:0})")
          session.execute("UPDATE t SET v.first = null WHERE id = 0")
          rows = list(session.execute("SELECT * FROM t WHERE id=0"))
          assert listify(rows) == [[0, [None, 3, 0]]]
  
          rows = list(session.execute("SELECT v.first FROM t WHERE id=0"))
          assert listify(rows) == [[None]]
          rows = list(session.execute("SELECT v.second FROM t WHERE id=0"))
          assert listify(rows) == [[3]]
          rows = list(session.execute("SELECT v.third FROM t WHERE id=0"))
          assert listify(rows) == [[0]]
  
      @since('2.2')
      def test_user_type_isolation(self):
          """
          Ensure UDT cannot be used from another keyspace
          @jira_ticket CASSANDRA-9409
          @since 2.2
          """
          cluster = self.cluster
          cluster.populate(1).start()
          node1 = cluster.nodelist()[0]
          session = self.patient_cql_connection(node1)
          create_ks(session, 'user_types', 1)
  
          # create a user defined type in a keyspace
          session.execute("CREATE TYPE udt (first text, second int, third int)")
  
          # ensure we cannot use a udt from another keyspace
          create_ks(session, 'user_ks', 1)
          assert_invalid(
              session,
              "CREATE TABLE t (id int PRIMARY KEY, v frozen<user_types.udt>)",
              "Statement on keyspace user_ks cannot refer to a user type in keyspace user_types"
          )
  