PYTHON-1204: Make graph metadata handling more robust (scylladb#47)

* Make graph metadata handling more robust

Author: aboudreault

cassandra/metadata.py

@@ -2843,26 +2843,7 @@ def __init__(self, connection, timeout):
 
     def get_all_keyspaces(self):
         for keyspace_meta in super(SchemaParserDSE68, self).get_all_keyspaces():
-
-            def _build_table_graph_metadata(table_meta):
-                for row in self.keyspace_table_vertex_rows[keyspace_meta.name][table_meta.name]:
-                    vertex_meta = self._build_table_vertex_metadata(row)
-                    table_meta.vertex = vertex_meta
-
-                for row in self.keyspace_table_edge_rows[keyspace_meta.name][table_meta.name]:
-                    edge_meta = self._build_table_edge_metadata(keyspace_meta, row)
-                    table_meta.edge = edge_meta
-
-            # Make sure we process vertices before edges
-            for t in [t for t in six.itervalues(keyspace_meta.tables)
-                      if t.name in self.keyspace_table_vertex_rows[keyspace_meta.name]]:
-                _build_table_graph_metadata(t)
-
-            # all other tables...
-            for t in [t for t in six.itervalues(keyspace_meta.tables)
-                      if t.name not in self.keyspace_table_vertex_rows[keyspace_meta.name]]:
-                _build_table_graph_metadata(t)
-
+            self._build_graph_metadata(keyspace_meta)
             yield keyspace_meta
 
     def get_table(self, keyspaces, keyspace, table):
@@ -2877,10 +2858,16 @@ def get_table(self, keyspaces, keyspace, table):
         vertices_result = self._handle_results(vertices_success, vertices_result)
         edges_result = self._handle_results(edges_success, edges_result)
 
-        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])
+        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
 
@@ -2893,6 +2880,32 @@ def _build_keyspace_metadata_internal(row):
         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.egde = self._build_table_edge_metadata(keyspace_meta, row)
+
+        try:
+            # Make sure we process vertices before edges
+            for table_meta in [t for t in six.itervalues(keyspace_meta.tables)
+                               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 six.itervalues(keyspace_meta.tables)
+                               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 six.itervalues(keyspace_meta.tables):
+                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"),

tests/integration/advanced/test_adv_metadata.py

@@ -14,11 +14,12 @@
 
 from packaging.version import Version
 
+from cassandra.cluster import Cluster
 from tests.integration import (BasicExistingKeyspaceUnitTestCase, BasicSharedKeyspaceUnitTestCase,
                                BasicSharedKeyspaceUnitTestCaseRF1,
                                greaterthanorequaldse51, greaterthanorequaldse60,
                                greaterthanorequaldse68, use_single_node,
-                               DSE_VERSION, requiredse)
+                               DSE_VERSION, requiredse, PROTOCOL_VERSION)
 
 try:
     import unittest2 as unittest
@@ -361,3 +362,35 @@ def test_edge_metadata(self):
         self.assertEqual(edge_meta.to_label, 'rocksolidsoftware')
         self.assertEqual(edge_meta.to_partition_key_columns, ['company_name', 'software_name'])
         self.assertEqual(edge_meta.to_clustering_columns, ['software_version'])
+
+
+@greaterthanorequaldse68
+class GraphMetadataSchemaErrorTests(BasicExistingKeyspaceUnitTestCase):
+    """
+    Test that we can connect when the graph schema is broken.
+    """
+
+    def test_connection_on_graph_schema_error(self):
+        self.session = self.cluster.connect()
+
+        self.session.execute("""
+        CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1} and graph_engine = 'Core';
+        """ % (self.ks_name,))
+
+        self.session.execute("""
+        CREATE TABLE %s.person (name text PRIMARY KEY) WITH VERTEX LABEL;
+        """ % (self.ks_name,))
+
+        self.session.execute("""
+        CREATE TABLE %s.software(company  text, name text, version int, PRIMARY KEY((company, name), version)) WITH VERTEX LABEL rocksolidsoftware;
+        """ % (self.ks_name,))
+
+        self.session.execute("""
+        CREATE TABLE %s.contributors (contributor text, company_name text, software_name text, software_version int,
+        PRIMARY KEY (contributor, company_name, software_name, software_version) )
+        WITH CLUSTERING ORDER BY (company_name ASC, software_name ASC, software_version ASC)
+        AND EDGE LABEL contrib FROM person(contributor) TO rocksolidsoftware((company_name, software_name), software_version);
+        """ % (self.ks_name,))
+
+        self.session.execute('TRUNCATE system_schema.vertices')
+        Cluster(protocol_version=PROTOCOL_VERSION).connect().shutdown()