From 18e7ff12fde17ba2b6e21d5f4234c5ed207ce4af Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 11 Oct 2026 16:47:18 +0000 Subject: [PATCH] metadata: remove DSE-only schema parsing ScyllaDB never sets system.local.dse_version, so get_schema_parser() never selected the DataStax Enterprise (DSE) schema parsers against it, and the graph/RLAC metadata they produced could never be populated. The code was already excluded from coverage for that reason. This follows the removal of DSE Graph support (#1128). Removed from cassandra/metadata.py: - SchemaParserDSE60, SchemaParserDSE67, SchemaParserDSE68 and the `if dse_version:` selection block in get_schema_parser(). - TableMetadataDSE68 and its vertex/edge attributes, VertexMetadata and EdgeMetadata (DSE 6.8 graph schema from system_schema.vertices/edges), and the vertex-first table ordering in KeyspaceMetadata.export_as_string(). - KeyspaceMetadata.graph_engine, its constructor argument, the graph_engine clause in as_cql_query(), and is_graph_enabled. - RLACTableExtension (DSE_RLACA row-level access control extension). - The DSE 6.0 'nodesync' entry in TableMetadataV3.option_maps; only the DSE parsers ever put that option into table options. get_schema_parser() keeps its dse_version parameter so positional callers are not silently shifted, but now ignores it. When release_version is unknown it no longer issues a second system.local query for dse_version (which ScyllaDB and Cassandra reject). Metadata.refresh() passes None for it. Host.dse_version is untouched: cluster.py still reads it for DSE native_address handling. Tests: drop the DSE parser selection and graph CQL export tests, the simulacron priming of the vertices/edges queries, the integration 'graph' workload JVM sizing, and the RLAC entry in the integration extension registry count. Add generic TableMetadataV3 CQL export, registered table extension and qualified replication class tests, which the removed graph tests used to exercise indirectly. Also drop the removed classes from the API docs and their coverage exclusions in pyproject.toml, and add a CHANGELOG entry. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QYcPjKDEQaQTfxpdfsFjZA --- CHANGELOG.rst | 12 + cassandra/metadata.py | 394 +------------------- docs/api/cassandra/metadata.rst | 9 - pyproject.toml | 3 - tests/integration/__init__.py | 7 +- tests/integration/simulacron/utils.py | 9 +- tests/integration/standard/test_metadata.py | 3 +- tests/unit/advanced/test_metadata.py | 99 +---- tests/unit/test_metadata.py | 116 +++--- 9 files changed, 90 insertions(+), 562 deletions(-) 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): """