Make sure to only query the native_transport_address column with DSE

Author: aboudreault

CHANGELOG.rst

@@ -6,6 +6,10 @@ Features
 --------
 * Add all() function to the ResultSet API (PYTHON-1203)
 
+Bug Fixes
+---------
+* Make sure to only query the native_transport_address column with DSE (PYTHON-1205)
+
 3.21.0
 ======
 January 15, 2020

build.yaml

@@ -9,7 +9,7 @@ schedules:
     matrix:
       exclude:
         - python: [3.4, 3.6, 3.7, 3.8]
-        - cassandra: ['2.1', '3.0', 'test-dse']
+        - cassandra: ['2.1', '3.0', '4.0', 'test-dse']
 
   commit_long_test:
     schedule: per_commit
@@ -21,7 +21,7 @@ schedules:
     matrix:
       exclude:
         - python: [3.4, 3.6, 3.7, 3.8]
-        - cassandra: ['2.1', '3.0', 'test-dse']
+        - cassandra: ['2.1', '3.0', '4.0', 'test-dse']
 
   commit_branches:
     schedule: per_commit
@@ -34,7 +34,7 @@ schedules:
     matrix:
       exclude:
         - python: [3.4, 3.6, 3.7, 3.8]
-        - cassandra: ['2.1', '3.0', 'test-dse']
+        - cassandra: ['2.1', '3.0', '4.0', 'test-dse']
 
   commit_branches_dev:
     schedule: per_commit
@@ -47,7 +47,7 @@ schedules:
     matrix:
       exclude:
         - python: [2.7, 3.4, 3.7, 3.6, 3.8]
-        - cassandra: ['2.0', '2.1', '2.2', '3.0', '3.11', 'test-dse', 'dse-4.8', 'dse-5.0', 'dse-6.0', 'dse-6.7', 'dse-6.8']
+        - cassandra: ['2.0', '2.1', '2.2', '3.0', '4.0', 'test-dse', 'dse-4.8', 'dse-5.0', 'dse-6.0', 'dse-6.8']
 
   release_test:
     schedule: per_commit
@@ -139,7 +139,7 @@ schedules:
     matrix:
       exclude:
         - python: [3.4, 3.6, 3.7, 3.8]
-        - cassandra: ['2.0', '2.1', '2.2', '3.0', 'test-dse']
+        - cassandra: ['2.0', '2.1', '2.2', '3.0', '4.0', 'test-dse']
 
 python:
   - 2.7
@@ -182,7 +182,7 @@ build:
       pip install --upgrade pip
       pip install -U setuptools
 
-      pip install git+ssh://[email protected]/riptano/ccm-private.git
+      pip install $HOME/ccm
 
       if [ -n "$CCM_IS_DSE" ]; then
         pip install -r test-datastax-requirements.txt

cassandra/cluster.py

@@ -80,7 +80,7 @@
 from cassandra.marshal import int64_pack
 from cassandra.timestamps import MonotonicTimestampGenerator
 from cassandra.compat import Mapping
-from cassandra.util import _resolve_contact_points_to_string_map
+from cassandra.util import _resolve_contact_points_to_string_map, Version
 
 from cassandra.datastax.insights.reporter import MonitorReporter
 from cassandra.datastax.insights.util import version_supports_insights
@@ -3324,7 +3324,7 @@ class ControlConnection(object):
     _SELECT_SCHEMA_PEERS_TEMPLATE = "SELECT peer, host_id, {nt_col_name}, schema_version FROM system.peers"
     _SELECT_SCHEMA_LOCAL = "SELECT schema_version FROM system.local WHERE key='local'"
 
-    _MINIMUM_NATIVE_ADDRESS_VERSION = "4.0"
+    _MINIMUM_NATIVE_ADDRESS_DSE_VERSION = Version("6.0.0")
 
     _is_shutdown = False
     _timeout = None
@@ -3884,14 +3884,17 @@ def _peers_query_for_version(self, connection, peers_query_template):
         field named nt_col_name.
         """
         host_release_version = self._cluster.metadata.get_host(connection.endpoint).release_version
-        if host_release_version:
-            use_native_address_query = host_release_version >= self._MINIMUM_NATIVE_ADDRESS_VERSION
-            if use_native_address_query:
-                select_peers_query = peers_query_template.format(nt_col_name="native_transport_address")
-            else:
+        host_dse_version = self._cluster.metadata.get_host(connection.endpoint).dse_version
+        uses_native_address_query = (
+            host_dse_version and Version(host_dse_version) >= self._MINIMUM_NATIVE_ADDRESS_DSE_VERSION)
+
+        if uses_native_address_query:
+            select_peers_query = peers_query_template.format(nt_col_name="native_transport_address")
+        elif host_release_version:
                 select_peers_query = peers_query_template.format(nt_col_name="rpc_address")
         else:
             select_peers_query = self._SELECT_PEERS
+
         return select_peers_query
 
     def _signal_error(self):

tests/integration/__init__.py

@@ -172,7 +172,7 @@ def _get_dse_version_from_cass(cass_version):
         cassandra_version = Version(mcv_string)
 
     CASSANDRA_VERSION = Version(mcv_string) if mcv_string else cassandra_version
-    CCM_VERSION = mcv_string or cv_string
+    CCM_VERSION = mcv_string if mcv_string else cv_string
 
 CASSANDRA_IP = os.getenv('CLUSTER_IP', '127.0.0.1')
 CASSANDRA_DIR = os.getenv('CASSANDRA_DIR', None)
@@ -454,7 +454,7 @@ def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None,
     set_default_cass_ip()
 
     if ccm_options is None and DSE_VERSION:
-        ccm_options = {"version": DSE_VERSION}
+        ccm_options = {"version": CCM_VERSION}
     elif ccm_options is None:
         ccm_options = CCM_KWARGS.copy()