| import pytest |
| import logging |
| |
| from dtest import RUN_STATIC_UPGRADE_MATRIX, Tester |
| from distutils.version import LooseVersion |
| from tools.misc import add_skip |
| from .upgrade_manifest import build_upgrade_pairs, CASSANDRA_3_0 |
| |
| since = pytest.mark.since |
| logger = logging.getLogger(__name__) |
| |
| @pytest.mark.upgrade_test |
| @since('3.11.6') |
| class UpgradeUDTFixTest(Tester): |
| __test__ = False |
| |
| """ |
| @jira_ticket: CASSANDRA-TBD |
| |
| 3.0 sstable cannot be read by 3.11 and newer |
| |
| OSS C* 3.0 only support frozen UDTs. A frozen type is marked using |
| "org.apache.cassandra.db.marshal.FrozenType(...)" using the AbstractType class hierarchy. |
| But 3.0 does not write the "FrozenType()" type into the serialization header in the sstable |
| metadata. CASSANDRA-7423 added support for non-frozen UDTs) in C* 3.6. Since then, UDTs can |
| be either frozen or non-frozen, so it becomes important that the "FrozenType bracket" is |
| present in the serialization header, because the serialization order of columns depends on |
| its type (fixed length type column first, then variable length type columns, then multi-cell |
| (non-frozen) type columns). This means, that C* 3.6 and newer will interpret the originally |
| frozen UDT as non-frozen and deserialization fails. |
| """ |
| |
| def test_udtfix_in_sstable(self): |
| """ |
| Verify that the fix to modify the serialization-header in the sstable metadata works for a single node. |
| """ |
| cluster = self.cluster |
| cluster.set_install_dir(version=self.UPGRADE_PATH.starting_version) |
| cluster.populate(1) |
| node1, = cluster.nodelist() |
| cluster.start() |
| |
| session = self.patient_cql_connection(node1) |
| self._schema_and_data(node1, session, "ks", "{'class': 'SimpleStrategy', 'replication_factor': 1}") |
| references = self._read_rows(session, "ks") |
| session.cluster.shutdown() |
| |
| logger.debug("Upgrading from {} to {}".format(self.UPGRADE_PATH.starting_version, self.UPGRADE_PATH.upgrade_version)) |
| cluster.stop() |
| cluster.set_install_dir(version=self.UPGRADE_PATH.upgrade_version) |
| |
| log_mark = node1.mark_log() |
| cluster.start() |
| self._verify_upgrade_log(log_mark, node1) |
| |
| session = self.patient_cql_connection(node1) |
| check = self._read_rows(session, "ks") |
| assert references == check |
| |
| def test_udtfix_in_messaging(self): |
| """ |
| Paranoia test for CASSANDRA-TBD, but verify that internode reads are not affected - with the data |
| both on the upgraded and non-upgraded node. |
| """ |
| cluster = self.cluster |
| cluster.set_install_dir(version=self.UPGRADE_PATH.starting_version) |
| cluster.populate([1, 1]) |
| node1, node2 = cluster.nodelist() |
| cluster.start() |
| |
| logger.debug("Started 2-DC cluster") |
| |
| session = self.patient_cql_connection(node1) |
| self._schema_and_data(node1, session, "ks1", "{'class': 'NetworkTopologyStrategy', 'dc1': 1}") |
| self._schema_and_data(node1, session, "ks2", "{'class': 'NetworkTopologyStrategy', 'dc2': 1}") |
| references11 = self._read_rows(session, "ks1") |
| references12 = self._read_rows(session, "ks2") |
| session.cluster.shutdown() |
| |
| session = self.patient_cql_connection(node2) |
| references21 = self._read_rows(session, "ks1") |
| references22 = self._read_rows(session, "ks2") |
| session.cluster.shutdown() |
| |
| assert references11 == references21 |
| assert references12 == references22 |
| |
| logger.debug("Upgrading {} from {} to {}".format(node1.name, self.UPGRADE_PATH.starting_version, self.UPGRADE_PATH.upgrade_version)) |
| node1.stop() |
| node1.set_install_dir(version=self.UPGRADE_PATH.upgrade_version) |
| cluster.set_configuration_options() # re-write cassandra-topology.properties (it's overridden by Node.set_install_dir) |
| log_mark = node1.mark_log() |
| node1.start(wait_for_binary_proto=True) |
| self._verify_upgrade_log(log_mark, node1) |
| |
| session = self.exclusive_cql_connection(node1) |
| check11 = self._read_rows(session, "ks1") |
| check12 = self._read_rows(session, "ks2") |
| assert references11 == check11 |
| assert references12 == check12 |
| session.cluster.shutdown() |
| |
| session = self.exclusive_cql_connection(node2) |
| check21 = self._read_rows(session, "ks1") |
| check22 = self._read_rows(session, "ks2") |
| assert references11 == check21 |
| assert references12 == check22 |
| session.cluster.shutdown() |
| |
| def _schema_and_data(self, node, session, ks, replication): |
| logger.debug("Creating schema and writing data (+flush)") |
| session.execute("CREATE KEYSPACE {} " |
| "WITH replication = {}".format(ks, replication)) |
| session.execute("CREATE TYPE {}.user_type (some_field text, some_int int)".format(ks)) |
| session.execute("CREATE TABLE {}.some_table (id int PRIMARY KEY, " |
| "a_udt_column_name frozen<user_type>, " |
| "col_a int, " |
| "col_c int)".format(ks)) |
| # NOTE: the names (internal order) of the regular columns matters (for this test)! |
| # Regular column order is: |
| # 1. fixed length type columns |
| # 2. variable length type columns |
| # 3. complex/multi-cell type columns |
| # The UDT field must not be the last field. |
| session.execute("INSERT INTO {}.some_table (id, col_a, a_udt_column_name, col_c) " |
| "VALUES (1, 12, {{some_field:'some', some_int:42}}, 13)".format(ks)) |
| session.execute("INSERT INTO {}.some_table (id, col_a, a_udt_column_name, col_c) " |
| "VALUES (2, 14, {{some_field:'field', some_int:11}}, 15)".format(ks)) |
| session.execute("INSERT INTO {}.some_table (id, col_c) " |
| "VALUES (3, 16)".format(ks)) |
| session.execute("INSERT INTO {}.some_table (id, col_c) " |
| "VALUES (4, 17)".format(ks)) |
| node.flush() |
| |
| def _read_rows(self, session, ks): |
| return [list(session.execute("SELECT id, col_a, a_udt_column_name, col_c " |
| "FROM {}.some_table WHERE id = {}".format(ks, x + 1)))[0] for x in range(4)] |
| |
| def _verify_upgrade_log(self, log_mark, node): |
| logger.debug("Verifying messages in log file") |
| assert 1 == len(node.grep_log( |
| 'Detected upgrade from .* to .*, fixing UDT type references in sstable metadata serialization-headers', |
| from_mark=log_mark)) |
| assert 1 == len(node.grep_log("Column 'a_udt_column_name' needs to be updated from type " |
| "'user_type' to 'frozen<user_type>'", from_mark=log_mark)) |
| assert 1 == len(node.grep_log('Writing new metadata file ', from_mark=log_mark)) |
| assert 1 == len(node.grep_log('Finished scanning all data directories...', from_mark=log_mark)) |
| |
| |
| for path in build_upgrade_pairs(): |
| gen_class_name = UpgradeUDTFixTest.__name__ + '_' + path.name |
| assert gen_class_name not in globals() |
| spec = {'UPGRADE_PATH': path, |
| '__test__': True} |
| |
| start_family = spec['UPGRADE_PATH'].starting_meta.family |
| upgrade_family = spec['UPGRADE_PATH'].upgrade_meta.family |
| start_family_applies = start_family == CASSANDRA_3_0 |
| upgrade_applies_to_env = RUN_STATIC_UPGRADE_MATRIX or start_family_applies |
| cls = type(gen_class_name, (UpgradeUDTFixTest,), spec) |
| if not upgrade_applies_to_env: |
| add_skip(cls, 'test not applicable to env.') |
| globals()[gen_class_name] = cls |
