Pull version information from systel.local, when version info is not

present

Author: dkropachev

cassandra/metadata.py

@@ -139,8 +139,9 @@ def export_schema_as_string(self):
     def refresh(self, connection, timeout, target_type=None, change_type=None, fetch_size=None,
                 metadata_request_timeout=None, **kwargs):
 
-        server_version = self.get_host(connection.original_endpoint).release_version
-        dse_version = self.get_host(connection.original_endpoint).dse_version
+        host = 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)
 
         if not target_type:
@@ -3410,7 +3411,32 @@ def __init__(
 
 
 def get_schema_parser(connection, server_version, dse_version, timeout, metadata_request_timeout, fetch_size=None):
-    version = Version(server_version)
+    if server_version is None and dse_version is None:
+        success, local_result = connection.wait_for_response(
+            QueryMessage(
+                query=maybe_add_timeout_to_query(
+                    "SELECT release_version FROM system.local WHERE key='local'",
+                    metadata_request_timeout),
+                consistency_level=ConsistencyLevel.ONE)
+            , timeout=timeout, fail_on_error=False)
+        if success and local_result.parsed_rows:
+            local_rows = dict_factory(local_result.column_names, local_result.parsed_rows)
+            local_row = local_rows[0]
+            server_version = local_row.get("release_version")
+
+        success, local_result = connection.wait_for_response(
+            QueryMessage(
+                query=maybe_add_timeout_to_query(
+                    "SELECT dse_version FROM system.local WHERE key='local'",
+                    metadata_request_timeout),
+                consistency_level=ConsistencyLevel.ONE)
+            , timeout=timeout, fail_on_error=False)
+        if success and local_result.parsed_rows:
+            local_rows = dict_factory(local_result.column_names, local_result.parsed_rows)
+            local_row = local_rows[0]
+            server_version = local_row.get("dse_version")
+
+    version = Version(server_version or "0")
     if dse_version:
         v = Version(dse_version)
         if v >= Version('6.8.0'):

tests/unit/test_metadata.py

@@ -30,7 +30,8 @@
                                 UserType, KeyspaceMetadata, get_schema_parser,
                                 _UnknownStrategy, ColumnMetadata, TableMetadata,
                                 IndexMetadata, Function, Aggregate,
-                                Metadata, TokenMap, ReplicationFactor)
+                                Metadata, TokenMap, ReplicationFactor,
+                                SchemaParserDSE68)
 from cassandra.policies import SimpleConvictionPolicy
 from cassandra.pool import Host
 from tests.util import assertCountEqual
@@ -616,6 +617,22 @@ def test_build_index_as_cql(self):
         assert index_meta.as_cql_query() == "CREATE CUSTOM INDEX index_name_here ON keyspace_name_here.table_name_here (column_name_here) USING 'class_name_here'"
 
 
+class SchemaParserLookupTests(unittest.TestCase):
+
+    def test_reads_versions_from_system_local_when_missing(self):
+        connection = Mock()
+        local_result = Mock()
+        local_result.column_names = ["release_version", "dse_version"]
+        local_result.parsed_rows = [["4.0.0", "6.8.0"]]
+        connection.wait_for_response.return_value = (True, local_result)
+
+        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):
     """
     Exercise cql generation with unicode characters. Keyspace, Table, and Index names