diff --git a/CHANGELOG.rst b/CHANGELOG.rst index a0d20f0d31..725d6ea05c 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -78,6 +78,18 @@ Behavior Changes timing out on the client. With ``raise_on_first_error=True``, both result modes now stop scheduling new statements after the first failure. +Removed +------- +* DataStax Enterprise (DSE)-only schema parsing has been removed: + ``SchemaParserDSE60``, ``SchemaParserDSE67``, ``SchemaParserDSE68``, + ``TableMetadataDSE68``, ``VertexMetadata``, ``EdgeMetadata``, + ``KeyspaceMetadata.graph_engine`` (including the ``graph_engine`` + constructor argument and ``KeyspaceMetadata.is_graph_enabled``) and + ``RLACTableExtension``. ScyllaDB never reports ``dse_version``, so these were + never used against it. The ``dse_version`` argument of + ``get_schema_parser()`` is now ignored, and the driver no longer queries + ``system.local.dse_version`` when ``release_version`` is unknown. + 3.29.12 ======== Sep 26, 2026 diff --git a/cassandra/metadata.py b/cassandra/metadata.py index a4eb7c52cd..ba3b6778f9 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -146,8 +146,7 @@ def refresh(self, connection, timeout, target_type=None, change_type=None, fetch host = (self.get_host_by_host_id(host_id) if host_id is not None else self.get_host(connection.original_endpoint)) server_version = host.release_version if host else None - dse_version = host.dse_version if host else None - parser = get_schema_parser(connection, server_version, dse_version, timeout, metadata_request_timeout, fetch_size) + parser = get_schema_parser(connection, server_version, None, timeout, metadata_request_timeout, fetch_size) if not target_type: self._rebuild_all(parser) @@ -868,11 +867,6 @@ class KeyspaceMetadata(object): .. versionadded:: 3.15 """ - graph_engine = None - """ - A string indicating whether a graph engine is enabled for this keyspace (Core/Classic). - """ - _consistency_mode = _ConsistencyMode.EVENTUAL """ The consistency mode of the keyspace, derived from the the ``consistency`` @@ -885,7 +879,7 @@ class KeyspaceMetadata(object): _exc_info = None """ set if metadata parsing failed """ - def __init__(self, name, durable_writes, strategy_class, strategy_options, graph_engine=None): + def __init__(self, name, durable_writes, strategy_class, strategy_options): self.name = name self.durable_writes = durable_writes self.replication_strategy = ReplicationStrategy.create(strategy_class, strategy_options) @@ -895,28 +889,19 @@ def __init__(self, name, durable_writes, strategy_class, strategy_options, graph self.functions = {} self.aggregates = {} self.views = {} - self.graph_engine = graph_engine self._consistency_mode = _ConsistencyMode.EVENTUAL - @property - def is_graph_enabled(self): - return self.graph_engine is not None - def export_as_string(self): """ Returns a CQL query string that can be used to recreate the entire keyspace, including user-defined types and tables. """ - # Make sure tables with vertex are exported before tables with edges - tables_with_vertex = [t for t in self.tables.values() if hasattr(t, 'vertex') and t.vertex] - other_tables = [t for t in self.tables.values() if t not in tables_with_vertex] - cql = "\n\n".join( [self.as_cql_query() + ';'] + self.user_type_strings() + [f.export_as_string() for f in self.functions.values()] + [a.export_as_string() for a in self.aggregates.values()] + - [t.export_as_string() for t in tables_with_vertex + other_tables]) + [t.export_as_string() for t in self.tables.values()]) if self._exc_info: import traceback @@ -953,8 +938,6 @@ def as_cql_query(self): ret = ret + (" AND consistency = '%s'" % self._consistency_mode.value) ret = ret + (' AND durable_writes = %s' % ("true" if self.durable_writes else "false")) - if self.graph_engine is not None: - ret = ret + (" AND graph_engine = '%s'" % self.graph_engine) return ret def user_type_strings(self): @@ -1587,10 +1570,7 @@ class TableMetadataV3(TableMetadata): """ compaction_options = {} - option_maps = [ - 'compaction', 'compression', 'caching', - 'nodesync' # added DSE 6.0 - ] + option_maps = ['compaction', 'compression', 'caching'] @property def is_cql_compatible(self): @@ -1617,52 +1597,6 @@ def _make_option_strings(cls, options_map): return list(sorted(ret)) -class TableMetadataDSE68(TableMetadataV3): - - vertex = None - """A :class:`.VertexMetadata` instance, if graph enabled""" - - edge = None - """A :class:`.EdgeMetadata` instance, if graph enabled""" - - def as_cql_query(self, formatted=False): - ret = super(TableMetadataDSE68, self).as_cql_query(formatted) - - if self.vertex: - ret += " AND VERTEX LABEL %s" % protect_name(self.vertex.label_name) - - if self.edge: - ret += " AND EDGE LABEL %s" % protect_name(self.edge.label_name) - - ret += self._export_edge_as_cql( - self.edge.from_label, - self.edge.from_partition_key_columns, - self.edge.from_clustering_columns, "FROM") - - ret += self._export_edge_as_cql( - self.edge.to_label, - self.edge.to_partition_key_columns, - self.edge.to_clustering_columns, "TO") - - return ret - - @staticmethod - def _export_edge_as_cql(label_name, partition_keys, - clustering_columns, keyword): - ret = " %s %s(" % (keyword, protect_name(label_name)) - - if len(partition_keys) == 1: - ret += protect_name(partition_keys[0]) - else: - ret += "(%s)" % ", ".join([protect_name(k) for k in partition_keys]) - - if clustering_columns: - ret += ", %s" % ", ".join([protect_name(k) for k in clustering_columns]) - ret += ")" - - return ret - - class TableExtensionInterface(object): """ Defines CQL/DDL for Cassandra table extensions. @@ -3064,14 +2998,6 @@ def _schema_type_to_cql(type_string): return type_string -class SchemaParserDSE60(SchemaParserV3): - """ - For DSE 6.0+ - """ - recognized_table_options = (SchemaParserV3.recognized_table_options + - ("nodesync",)) - - class SchemaParserV4(SchemaParserV3): recognized_table_options = ( @@ -3235,213 +3161,6 @@ def _build_keyspace_metadata_internal(row): return super(SchemaParserV4, SchemaParserV4)._build_keyspace_metadata_internal(row) -class SchemaParserDSE67(SchemaParserV4): - """ - For DSE 6.7+ - """ - recognized_table_options = (SchemaParserV4.recognized_table_options + - ("nodesync",)) - - -class SchemaParserDSE68(SchemaParserDSE67): - """ - For DSE 6.8+ - """ - - _SELECT_VERTICES = "SELECT * FROM system_schema.vertices" - _SELECT_EDGES = "SELECT * FROM system_schema.edges" - - _table_metadata_class = TableMetadataDSE68 - - def __init__(self, connection, timeout, fetch_size, metadata_request_timeout): - super(SchemaParserDSE68, self).__init__(connection, timeout, fetch_size, metadata_request_timeout) - self.keyspace_table_vertex_rows = defaultdict(lambda: defaultdict(list)) - self.keyspace_table_edge_rows = defaultdict(lambda: defaultdict(list)) - - def get_all_keyspaces(self): - for keyspace_meta in super(SchemaParserDSE68, self).get_all_keyspaces(): - self._build_graph_metadata(keyspace_meta) - yield keyspace_meta - - def get_table(self, keyspaces, keyspace, table): - table_meta = super(SchemaParserDSE68, self).get_table(keyspaces, keyspace, table) - cl = ConsistencyLevel.ONE - where_clause = bind_params(" WHERE keyspace_name = %%s AND %s = %%s" % (self._table_name_col), (keyspace, table), _encoder) - vertices_query = QueryMessage( - query=maybe_add_timeout_to_query(self._SELECT_VERTICES + where_clause, self.metadata_request_timeout), - consistency_level=cl, - ) - edges_query = QueryMessage( - query=maybe_add_timeout_to_query(self._SELECT_EDGES + where_clause, self.metadata_request_timeout), - consistency_level=cl, - ) - - (vertices_success, vertices_result), (edges_success, edges_result) \ - = self.connection.wait_for_responses(vertices_query, edges_query, timeout=self.timeout, fail_on_error=False) - vertices_result = self._handle_results(vertices_success, vertices_result) - edges_result = self._handle_results(edges_success, edges_result) - - try: - if vertices_result: - table_meta.vertex = self._build_table_vertex_metadata(vertices_result[0]) - elif edges_result: - table_meta.edge = self._build_table_edge_metadata(keyspaces[keyspace], edges_result[0]) - except Exception: - table_meta.vertex = None - table_meta.edge = None - table_meta._exc_info = sys.exc_info() - log.exception("Error while parsing graph metadata for table %s.%s.", keyspace, table) - - return table_meta - - @staticmethod - def _build_keyspace_metadata_internal(row): - name = row["keyspace_name"] - durable_writes = row.get("durable_writes", None) - replication = dict(row.get("replication")) if 'replication' in row else {} - replication_class = replication.pop("class") if 'class' in replication else None - graph_engine = row.get("graph_engine", None) - return KeyspaceMetadata(name, durable_writes, replication_class, replication, graph_engine) - - def _build_graph_metadata(self, keyspace_meta): - - def _build_table_graph_metadata(table_meta): - for row in self.keyspace_table_vertex_rows[keyspace_meta.name][table_meta.name]: - table_meta.vertex = self._build_table_vertex_metadata(row) - - for row in self.keyspace_table_edge_rows[keyspace_meta.name][table_meta.name]: - table_meta.edge = self._build_table_edge_metadata(keyspace_meta, row) - - try: - # Make sure we process vertices before edges - for table_meta in [t for t in keyspace_meta.tables.values() - if t.name in self.keyspace_table_vertex_rows[keyspace_meta.name]]: - _build_table_graph_metadata(table_meta) - - # all other tables... - for table_meta in [t for t in keyspace_meta.tables.values() - if t.name not in self.keyspace_table_vertex_rows[keyspace_meta.name]]: - _build_table_graph_metadata(table_meta) - except Exception: - # schema error, remove all graph metadata for this keyspace - for t in keyspace_meta.tables.values(): - t.edge = t.vertex = None - keyspace_meta._exc_info = sys.exc_info() - log.exception("Error while parsing graph metadata for keyspace %s", keyspace_meta.name) - - @staticmethod - def _build_table_vertex_metadata(row): - return VertexMetadata(row.get("keyspace_name"), row.get("table_name"), - row.get("label_name")) - - @staticmethod - def _build_table_edge_metadata(keyspace_meta, row): - from_table = row.get("from_table") - from_table_meta = keyspace_meta.tables.get(from_table) - from_label = from_table_meta.vertex.label_name - to_table = row.get("to_table") - to_table_meta = keyspace_meta.tables.get(to_table) - to_label = to_table_meta.vertex.label_name - - return EdgeMetadata( - row.get("keyspace_name"), row.get("table_name"), - row.get("label_name"), from_table, from_label, - row.get("from_partition_key_columns"), - row.get("from_clustering_columns"), to_table, to_label, - row.get("to_partition_key_columns"), - row.get("to_clustering_columns")) - - def _query_all(self): - cl = ConsistencyLevel.ONE - queries = [ - # copied from v4 - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_KEYSPACES, self.metadata_request_timeout), - consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_TABLES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_COLUMNS, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_TYPES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_FUNCTIONS, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_AGGREGATES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_TRIGGERS, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_INDEXES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_VIEWS, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_VIRTUAL_KEYSPACES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_VIRTUAL_TABLES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_VIRTUAL_COLUMNS, self.metadata_request_timeout), consistency_level=cl), - # dse6.8 only - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_VERTICES, self.metadata_request_timeout), consistency_level=cl), - QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_EDGES, self.metadata_request_timeout), consistency_level=cl) - ] - - responses = self.connection.wait_for_responses( - *queries, timeout=self.timeout, fail_on_error=False) - ( - # copied from V4 - (ks_success, ks_result), - (table_success, table_result), - (col_success, col_result), - (types_success, types_result), - (functions_success, functions_result), - (aggregates_success, aggregates_result), - (triggers_success, triggers_result), - (indexes_success, indexes_result), - (views_success, views_result), - (virtual_ks_success, virtual_ks_result), - (virtual_table_success, virtual_table_result), - (virtual_column_success, virtual_column_result), - # dse6.8 responses - (vertices_success, vertices_result), - (edges_success, edges_result) - ) = responses - - # copied from V4 - self.keyspaces_result = self._handle_results(ks_success, ks_result) - self.tables_result = self._handle_results(table_success, table_result) - self.columns_result = self._handle_results(col_success, col_result) - self.triggers_result = self._handle_results(triggers_success, triggers_result) - self.types_result = self._handle_results(types_success, types_result) - self.functions_result = self._handle_results(functions_success, functions_result) - self.aggregates_result = self._handle_results(aggregates_success, aggregates_result) - self.indexes_result = self._handle_results(indexes_success, indexes_result) - self.views_result = self._handle_results(views_success, views_result) - - # These tables don't exist in some DSE versions reporting 4.X so we can - # ignore them if we got an error - self.virtual_keyspaces_result = self._handle_results( - virtual_ks_success, virtual_ks_result, - expected_failures=(InvalidRequest,) - ) - self.virtual_tables_result = self._handle_results( - virtual_table_success, virtual_table_result, - expected_failures=(InvalidRequest,) - ) - self.virtual_columns_result = self._handle_results( - virtual_column_success, virtual_column_result, - expected_failures=(InvalidRequest,) - ) - - # dse6.8-only results - self.vertices_result = self._handle_results(vertices_success, vertices_result) - self.edges_result = self._handle_results(edges_success, edges_result) - - self._aggregate_results() - - def _aggregate_results(self): - super(SchemaParserDSE68, self)._aggregate_results() - - m = self.keyspace_table_vertex_rows - for row in self.vertices_result: - ksname = row["keyspace_name"] - cfname = row['table_name'] - m[ksname][cfname].append(row) - - m = self.keyspace_table_edge_rows - for row in self.edges_result: - ksname = row["keyspace_name"] - cfname = row['table_name'] - m[ksname][cfname].append(row) - - class MaterializedViewMetadata(object): """ A representation of a materialized view on a table @@ -3550,82 +3269,6 @@ def export_as_string(self): return self.as_cql_query(formatted=True) + ";" -class VertexMetadata(object): - """ - A representation of a vertex on a table - """ - - keyspace_name = None - """ A string name of the keyspace. """ - - table_name = None - """ A string name of the table this vertex is on. """ - - label_name = None - """ A string name of the label of this vertex.""" - - def __init__(self, keyspace_name, table_name, label_name): - self.keyspace_name = keyspace_name - self.table_name = table_name - self.label_name = label_name - - -class EdgeMetadata(object): - """ - A representation of an edge on a table - """ - - keyspace_name = None - """A string name of the keyspace """ - - table_name = None - """A string name of the table this edge is on""" - - label_name = None - """A string name of the label of this edge""" - - from_table = None - """A string name of the from table of this edge (incoming vertex)""" - - from_label = None - """A string name of the from table label of this edge (incoming vertex)""" - - from_partition_key_columns = None - """The columns that match the partition key of the incoming vertex table.""" - - from_clustering_columns = None - """The columns that match the clustering columns of the incoming vertex table.""" - - to_table = None - """A string name of the to table of this edge (outgoing vertex)""" - - to_label = None - """A string name of the to table label of this edge (outgoing vertex)""" - - to_partition_key_columns = None - """The columns that match the partition key of the outgoing vertex table.""" - - to_clustering_columns = None - """The columns that match the clustering columns of the outgoing vertex table.""" - - def __init__( - self, keyspace_name, table_name, label_name, from_table, - from_label, from_partition_key_columns, from_clustering_columns, - to_table, to_label, to_partition_key_columns, - to_clustering_columns): - self.keyspace_name = keyspace_name - self.table_name = table_name - self.label_name = label_name - self.from_table = from_table - self.from_label = from_label - self.from_partition_key_columns = from_partition_key_columns - self.from_clustering_columns = from_clustering_columns - self.to_table = to_table - self.to_label = to_label - self.to_partition_key_columns = to_partition_key_columns - self.to_clustering_columns = to_clustering_columns - - def _query_column_from_system_local(connection, column_name, timeout, metadata_request_timeout): success, local_result = connection.wait_for_response( QueryMessage( @@ -3651,30 +3294,19 @@ def get_column_from_system_local(connection, column_name: str, timeout, metadata def get_schema_parser(connection, server_version, dse_version, timeout, metadata_request_timeout, fetch_size=None): """Select the schema parser for the server's release line. - Cassandra and Scylla schema families are selected by major version; DSE - families are selected by major and minor version. Prerelease maturity does - not imply the preceding schema family, so versions such as - ``6.8.0-SNAPSHOT`` and ``6.8.0.rc1`` correctly select DSE68. + Cassandra and Scylla schema families are selected by major version. + + ``dse_version`` is ignored: the DataStax Enterprise (DSE) schema parsers + were removed. It is kept so existing positional callers keep working. """ missing_release_version_reason = None - if not server_version and not dse_version: + if not server_version: server_version, error = _query_column_from_system_local( connection, "release_version", timeout, metadata_request_timeout) - dse_version = get_column_from_system_local(connection, "dse_version", timeout, metadata_request_timeout) if not server_version: missing_release_version_reason = error or "empty value" version = Version(server_version or "0.0") - if dse_version: - v = Version(dse_version) - dse_release = (v.major, v.minor) - if dse_release >= (6, 8): - return SchemaParserDSE68(connection, timeout, fetch_size, metadata_request_timeout) - elif dse_release >= (6, 7): - return SchemaParserDSE67(connection, timeout, fetch_size, metadata_request_timeout) - elif dse_release >= (6, 0): - return SchemaParserDSE60(connection, timeout, fetch_size, metadata_request_timeout) - if version.major >= 4: return SchemaParserV4(connection, timeout, fetch_size, metadata_request_timeout) elif version.major >= 3: @@ -3701,14 +3333,6 @@ def _cql_from_cass_type(cass_type): return cass_type.cql_parameterized_type() -class RLACTableExtension(RegisteredTableExtension): - name = "DSE_RLACA" - - @classmethod - def after_table_cql(cls, table_meta, ext_key, ext_blob): - return "RESTRICT ROWS ON %s.%s USING %s;" % (protect_name(table_meta.keyspace_name), - protect_name(table_meta.name), - protect_name(ext_blob.decode('utf-8'))) NO_VALID_REPLICA = object() diff --git a/docs/api/cassandra/metadata.rst b/docs/api/cassandra/metadata.rst index 25526f61ec..df1673696b 100644 --- a/docs/api/cassandra/metadata.rst +++ b/docs/api/cassandra/metadata.rst @@ -39,9 +39,6 @@ Schemas .. autoclass:: TableMetadataV3 () :members: -.. autoclass:: TableMetadataDSE68 () - :members: - .. autoclass:: ColumnMetadata () :members: @@ -51,12 +48,6 @@ Schemas .. autoclass:: MaterializedViewMetadata () :members: -.. autoclass:: VertexMetadata () - :members: - -.. autoclass:: EdgeMetadata () - :members: - Tokens and Ring Topology ------------------------ diff --git a/pyproject.toml b/pyproject.toml index e9f136a6db..01f9f60a2b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -171,9 +171,6 @@ exclude_also = [ "^ def (_handle_continuous_paging_first_response|cancel_continuous_paging|new_continuous_paging_session|remove_continuous_paging_session|error_all_cp_sessions|_write_paging_options)\\(", # cluster.py: Astra `cloud=` secure-connect-bundle setup "^ +if cloud is not None:", - # metadata.py: DSE schema parsers, DSE Graph vertex/edge metadata, DSE RLAC - "^class (SchemaParserDSE60|SchemaParserDSE67|SchemaParserDSE68|TableMetadataDSE68|VertexMetadata|EdgeMetadata|RLACTableExtension)\\b", - "^ +if dse_version:", ] [tool.coverage.html] diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py index a8424bfc57..6eda8e9eb8 100644 --- a/tests/integration/__init__.py +++ b/tests/integration/__init__.py @@ -501,11 +501,8 @@ def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None, # This will enable the Mirroring query handler which will echo our custom payload k,v pairs back - if 'graph' in workloads: - jvm_args += ['-Xms1500M', '-Xmx1500M'] - else: - if PROTOCOL_VERSION >= 4 and not SCYLLA_VERSION: - jvm_args = [" -Dcassandra.custom_query_handler_class=org.apache.cassandra.cql3.CustomPayloadMirroringQueryHandler"] + if PROTOCOL_VERSION >= 4 and not SCYLLA_VERSION: + jvm_args = [" -Dcassandra.custom_query_handler_class=org.apache.cassandra.cql3.CustomPayloadMirroringQueryHandler"] if len(workloads) > 0: for node in CCM_CLUSTER.nodes.values(): node.set_workloads(workloads) diff --git a/tests/integration/simulacron/utils.py b/tests/integration/simulacron/utils.py index 283ec1b2e2..fa46edd432 100644 --- a/tests/integration/simulacron/utils.py +++ b/tests/integration/simulacron/utils.py @@ -16,7 +16,7 @@ import subprocess from urllib.request import build_opener, Request, HTTPHandler -from cassandra.metadata import SchemaParserV4, SchemaParserDSE68 +from cassandra.metadata import SchemaParserV4 from tests.util import wait_until_not_raised from tests.integration import CASSANDRA_VERSION, SIMULACRON_JAR @@ -390,13 +390,6 @@ def prime_driver_defaults(version=CASSANDRA_VERSION): "message": "Invalid Query!"}) ) - # prepare empty rows for NGDG - for query in [SchemaParserDSE68._SELECT_VERTICES, - SchemaParserDSE68._SELECT_EDGES]: - client_simulacron.submit_request( - PrimeQuery(query, result='success', - then={'rows': [], 'column_types': {'row1': 'int'}})) - def prime_cluster(data_centers="3", version=None, cluster_name=DEFAULT_CLUSTER, dse_version=None): """ diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py index 9b014d6112..dc4e3aa670 100644 --- a/tests/integration/standard/test_metadata.py +++ b/tests/integration/standard/test_metadata.py @@ -998,8 +998,7 @@ class Ext1(Ext0): assert Ext0.name in _RegisteredExtensionType._extension_registry assert Ext1.name in _RegisteredExtensionType._extension_registry - # There will bee the RLAC extension here. - assert len(_RegisteredExtensionType._extension_registry) == 3 + assert len(_RegisteredExtensionType._extension_registry) == 2 self.cluster.refresh_table_metadata(ks, t) table_meta = ks_meta.tables[t] diff --git a/tests/unit/advanced/test_metadata.py b/tests/unit/advanced/test_metadata.py index 1503759372..3450241034 100644 --- a/tests/unit/advanced/test_metadata.py +++ b/tests/unit/advanced/test_metadata.py @@ -14,107 +14,10 @@ import datetime import unittest -from cassandra.metadata import ( - KeyspaceMetadata, TableMetadataDSE68, - VertexMetadata, EdgeMetadata, _SchemaParser -) +from cassandra.metadata import _SchemaParser from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS -class GraphMetadataToCQLTests(unittest.TestCase): - - def _create_edge_metadata(self, partition_keys=['pk1'], clustering_keys=['c1']): - return EdgeMetadata( - 'keyspace', 'table', 'label', 'from_table', 'from_label', - partition_keys, clustering_keys, 'to_table', 'to_label', - partition_keys, clustering_keys) - - def _create_vertex_metadata(self, label_name='label'): - return VertexMetadata('keyspace', 'table', label_name) - - def _create_keyspace_metadata(self, graph_engine): - return KeyspaceMetadata( - 'keyspace', True, 'org.apache.cassandra.locator.NetworkTopologyStrategy', - {'dc1': 1}, graph_engine=graph_engine) - - def _create_table_metadata(self, with_vertex=False, with_edge=False): - tm = TableMetadataDSE68('keyspace', 'table') - if with_vertex: - tm.vertex = self._create_vertex_metadata() if with_vertex is True else with_vertex - elif with_edge: - tm.edge = self._create_edge_metadata() if with_edge is True else with_edge - - return tm - - def test_keyspace_no_graph_engine(self): - km = self._create_keyspace_metadata(None) - assert km.graph_engine == None - assert "graph_engine" not in km.as_cql_query() - - def test_keyspace_with_graph_engine(self): - graph_engine = 'Core' - km = self._create_keyspace_metadata(graph_engine) - assert km.graph_engine == graph_engine - cql = km.as_cql_query() - assert "graph_engine" in cql - assert "Core" in cql - - def test_table_no_vertex_or_edge(self): - tm = self._create_table_metadata() - assert tm.vertex is None - assert tm.edge is None - cql = tm.as_cql_query() - assert "VERTEX LABEL" not in cql - assert "EDGE LABEL" not in cql - - def test_table_with_vertex(self): - tm = self._create_table_metadata(with_vertex=True) - assert isinstance(tm.vertex, VertexMetadata) - assert tm.edge is None - cql = tm.as_cql_query() - assert "VERTEX LABEL" in cql - assert "EDGE LABEL" not in cql - - def test_table_with_edge(self): - tm = self._create_table_metadata(with_edge=True) - assert tm.vertex is None - assert isinstance(tm.edge, EdgeMetadata) - cql = tm.as_cql_query() - assert "VERTEX LABEL" not in cql - assert "EDGE LABEL" in cql - assert "FROM from_label" in cql - assert "TO to_label" in cql - - def test_vertex_with_label(self): - tm = self. _create_table_metadata(with_vertex=True) - assert tm.as_cql_query().endswith('VERTEX LABEL label') - - def test_edge_single_partition_key_and_clustering_key(self): - tm = self._create_table_metadata(with_edge=True) - assert 'FROM from_label(pk1, c1)' in tm.as_cql_query() - - def test_edge_multiple_partition_keys(self): - edge = self._create_edge_metadata(partition_keys=['pk1', 'pk2']) - tm = self. _create_table_metadata(with_edge=edge) - assert 'FROM from_label((pk1, pk2), ' in tm.as_cql_query() - - def test_edge_no_clustering_keys(self): - edge = self._create_edge_metadata(clustering_keys=[]) - tm = self. _create_table_metadata(with_edge=edge) - assert 'FROM from_label(pk1) ' in tm.as_cql_query() - - def test_edge_multiple_clustering_keys(self): - edge = self._create_edge_metadata(clustering_keys=['c1', 'c2']) - tm = self. _create_table_metadata(with_edge=edge) - assert 'FROM from_label(pk1, c1, c2) ' in tm.as_cql_query() - - def test_edge_multiple_partition_and_clustering_keys(self): - edge = self._create_edge_metadata(partition_keys=['pk1', 'pk2'], - clustering_keys=['c1', 'c2']) - tm = self. _create_table_metadata(with_edge=edge) - assert 'FROM from_label((pk1, pk2), c1, c2) ' in tm.as_cql_query() - - class SchemaParsersTests(unittest.TestCase): def test_metadata_query_metadata_timeout(self): class FakeConnection: diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index 3035fc8c60..62c6ac2288 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -33,13 +33,12 @@ _UnknownStrategy, ColumnMetadata, TableMetadata, IndexMetadata, Function, Aggregate, Metadata, TokenMap, ReplicationFactor, - SchemaParserDSE60, SchemaParserDSE67, SchemaParserDSE68, SchemaParserV22, SchemaParserV3, - SchemaParserV4, + SchemaParserV4, TableMetadataV3, + RegisteredTableExtension, _RegisteredExtensionType, _ConsistencyMode, _consistency_mode_from_string) from cassandra.policies import SimpleConvictionPolicy from cassandra.pool import Host -from cassandra.protocol import QueryMessage from cassandra.tablets import Tablet from tests.util import assertCountEqual import pytest @@ -620,6 +619,65 @@ def test_export_as_string_user_types(self): );""" == keyspace.export_as_string() +class TableMetadataV3ToCQLTest(unittest.TestCase): + + @staticmethod + def _table(options=None): + table = TableMetadataV3('ks', 'tbl', options=options) + pk = ColumnMetadata(table, 'k', 'int') + table.columns['k'] = pk + table.columns['v'] = ColumnMetadata(table, 'v', 'text') + table.partition_key.append(pk) + return table + + def test_as_cql_query_formats_map_and_scalar_options(self): + table = self._table(options={ + 'compaction': {'class': 'SizeTieredCompactionStrategy'}, + 'comment': None, + 'gc_grace_seconds': 864000, + }) + + assert table.as_cql_query() == ( + "CREATE TABLE ks.tbl (k int PRIMARY KEY, v text) WITH " + "compaction = {'class': 'SizeTieredCompactionStrategy'} " + "AND gc_grace_seconds = 864000") + + def test_as_cql_query_formatted(self): + assert self._table(options={'comment': 'c'}).as_cql_query(formatted=True) == ( + "CREATE TABLE ks.tbl (\n" + " k int PRIMARY KEY,\n" + " v text\n" + ") WITH comment = 'c'") + + def test_registered_extension_cql_follows_table(self): + class _TestExtension(RegisteredTableExtension): + name = 'unit_test_ext' + + @classmethod + def after_table_cql(cls, table_meta, ext_key, ext_blob): + return "-- %s %s %s;" % (table_meta.name, ext_key, ext_blob.decode('utf-8')) + + try: + assert _RegisteredExtensionType._extension_registry['unit_test_ext'] is _TestExtension + table = self._table() + table.extensions = {'unit_test_ext': b'blob', 'unregistered_ext': b'x'} + + exported = table.export_as_string() + + assert exported.endswith(";\n\n-- tbl unit_test_ext blob;") + assert 'unregistered_ext' not in exported + finally: + del _RegisteredExtensionType._extension_registry['unit_test_ext'] + + def test_keyspace_strips_qualified_strategy_class(self): + keyspace = KeyspaceMetadata( + 'ks', True, 'org.apache.cassandra.locator.NetworkTopologyStrategy', {'dc1': 1}) + + assert keyspace.as_cql_query() == ( + "CREATE KEYSPACE ks WITH replication = " + "{'class': 'NetworkTopologyStrategy', 'dc1': '1'} AND durable_writes = true") + + class KeyspaceConsistencyTabletInvalidationTest(unittest.TestCase): """ Metadata._update_keyspace must drop cached tablets when a keyspace's @@ -864,24 +922,6 @@ def test_scylla_release_versions_use_matching_parser(self): assert isinstance(parser, parser_class) - def test_dse_versions_use_matching_parser(self): - for dse_version, parser_class in ( - ('6.8.0', SchemaParserDSE68), - ('6.8.0-SNAPSHOT', SchemaParserDSE68), - ('6.8.0.rc1', SchemaParserDSE68), - ('6.7.0', SchemaParserDSE67), - ('6.7.0-SNAPSHOT', SchemaParserDSE67), - ('6.7.0.rc1', SchemaParserDSE67), - ('6.0.0', SchemaParserDSE60), - ('6.0.0-SNAPSHOT', SchemaParserDSE60), - ('6.0.0.rc1', SchemaParserDSE60), - ('5.1.20', SchemaParserV4)): - with self.subTest(dse_version=dse_version): - parser = get_schema_parser( - Mock(), '4.0.0', dse_version, 0.1, None) - - assert type(parser) is parser_class - def test_missing_versions_fall_back_to_v22_parser(self): connection = Mock() connection.wait_for_response.return_value = (False, None) @@ -903,8 +943,7 @@ def response(column_name, value): def mock_system_local(query, *args, **kwargs): if "release_version" in query.query: return (True, response("release_version", release_version)) - # Scylla and Cassandra reject the unknown dse_version column - return (False, Exception("Undefined column name dse_version")) + raise AssertionError("unexpected query: %s" % query.query) connection = Mock() connection.wait_for_response.side_effect = mock_system_local @@ -936,6 +975,8 @@ def test_scylla_release_version_from_system_local_uses_v3_parser(self): parser = get_schema_parser(connection, None, None, 0.1, None) assert type(parser) is SchemaParserV3 + # Only release_version is read; the DSE-only dse_version is not queried. + assert connection.wait_for_response.call_count == 1 def test_empty_release_version_is_read_from_system_local(self): connection = self._system_local_connection('3.0.8') @@ -981,35 +1022,6 @@ def test_refresh_uses_control_connection_host_id_for_versions(self): get_parser.assert_called_once_with( connection, None, None, 0.1, None, None) - def test_missing_release_version_with_dse_version_uses_dse_parser_without_warning(self): - connection = Mock() - - release_version_resp = Mock() - release_version_resp.column_names = ["release_version"] - release_version_resp.parsed_rows = [[None]] - - dse_version_resp = Mock() - dse_version_resp.column_names = ["dse_version"] - dse_version_resp.parsed_rows = [["6.8.0"]] - - def mock_system_local(query, *args, **kwargs): - if not isinstance(query, QueryMessage): - raise RuntimeError("first argument should be a QueryMessage") - if "release_version" in query.query: - return (True, release_version_resp) - if "dse_version" in query.query: - return (True, dse_version_resp) - raise RuntimeError("unexpected query") - - connection.wait_for_response.side_effect = mock_system_local - - with self.assertNoLogs('cassandra.metadata', level='WARNING'): - parser = get_schema_parser(connection, None, None, 0.1, None) - - assert isinstance(parser, SchemaParserDSE68) - message = connection.wait_for_response.call_args[0][0] - assert "system.local" in message.query - class UnicodeIdentifiersTests(unittest.TestCase): """