metadata: remove DSE-only schema parsing

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

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
 ------------------------
 

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]

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)

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):
     """

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]

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:

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):
     """