Merge pull request datastax#1067 from datastax/python-1205

PYTHON-1205: Fix Undefined column name native_transport_address" error when connecting to C* 4.0

Author: TheRealFalcon

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
@@ -46,8 +46,8 @@ schedules:
       EXCLUDE_LONG=1
     matrix:
       exclude:
-        - python: [2.7, 3.4, 3.7, 3.8]
-        - cassandra: ['2.0', '2.1', '2.2', '3.0', 'test-dse', dse-4.8', 'dse-5.0']
+        - python: [2.7, 3.4, 3.7, 3.6, 3.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
@@ -157,6 +157,7 @@ cassandra:
   - '2.2'
   - '3.0'
   - '3.11'
+  - '4.0'
   - 'dse-4.8'
   - 'dse-5.0'
   - 'dse-5.1'
@@ -181,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 = cassandra_version if mcv_string else CASSANDRA_VERSION
+    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,19 +454,13 @@ 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()
 
-    if 'version' in ccm_options and not isinstance(ccm_options['version'], Version):
-        ccm_options['version'] = Version(ccm_options['version'])
-
     cassandra_version = ccm_options.get('version', CCM_VERSION)
     dse_version = ccm_options.get('version', DSE_VERSION)
 
-    if 'version' in ccm_options:
-        ccm_options['version'] = ccm_options['version'].base_version
-
     global CCM_CLUSTER
     if USE_CASS_EXTERNAL:
         if CCM_CLUSTER:
@@ -515,12 +509,12 @@ def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None,
                 CCM_CLUSTER = DseCluster(path, cluster_name, **ccm_options)
                 CCM_CLUSTER.set_configuration_options({'start_native_transport': True})
                 CCM_CLUSTER.set_configuration_options({'batch_size_warn_threshold_in_kb': 5})
-                if dse_version >= Version('5.0'):
+                if Version(dse_version) >= Version('5.0'):
                     CCM_CLUSTER.set_configuration_options({'enable_user_defined_functions': True})
                     CCM_CLUSTER.set_configuration_options({'enable_scripted_user_defined_functions': True})
                 if 'spark' in workloads:
                     config_options = {"initial_spark_worker_resources": 0.1}
-                    if dse_version >= Version('6.7'):
+                    if Version(dse_version) >= Version('6.7'):
                         log.debug("Disabling AlwaysON SQL for a DSE 6.7 Cluster")
                         config_options['alwayson_sql_options'] = {'enabled': False}
                     CCM_CLUSTER.set_dse_configuration_options(config_options)
@@ -532,9 +526,9 @@ def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None,
             else:
                 CCM_CLUSTER = CCMCluster(path, cluster_name, **ccm_options)
                 CCM_CLUSTER.set_configuration_options({'start_native_transport': True})
-                if cassandra_version >= Version('2.2'):
+                if Version(cassandra_version) >= Version('2.2'):
                     CCM_CLUSTER.set_configuration_options({'enable_user_defined_functions': True})
-                    if cassandra_version >= Version('3.0'):
+                    if Version(cassandra_version) >= Version('3.0'):
                         CCM_CLUSTER.set_configuration_options({'enable_scripted_user_defined_functions': True})
                 common.switch_cluster(path, cluster_name)
                 CCM_CLUSTER.set_configuration_options(configuration_options)