[sql lab] allow EXPlAIN queries (#5558)

* [sql lab] allow EXPlAIN queries

closes andialbrecht/sqlparse#421

* typo

superset/sql_lab.py

@@ -130,7 +130,7 @@ def handle_error(msg):
     superset_query = SupersetQuery(rendered_query)
     executed_sql = superset_query.stripped()
     SQL_MAX_ROWS = app.config.get('SQL_MAX_ROW')
-    if not superset_query.is_select() and not database.allow_dml:
+    if not superset_query.is_readonly() and not database.allow_dml:
         return handle_error(
             'Only `SELECT` statements are allowed against this database')
     if query.select_as_cta:

superset/sql_parse.py

@@ -41,6 +41,13 @@ def limit(self):
     def is_select(self):
         return self._parsed[0].get_type() == 'SELECT'
 
+    def is_explain(self):
+        return self.sql.strip().upper().startswith('EXPLAIN')
+
+    def is_readonly(self):
+        """Pessimistic readonly, 100% sure statement won't mutate anything"""
+        return self.is_select() or self.is_explain()
+
     def stripped(self):
         return self.sql.strip(' \t\n;')
 

tests/sql_parse_tests.py

@@ -292,9 +292,21 @@ def test_reusing_aliases(self):
         """
         self.assertEquals({'src'}, self.extract_tables(query))
 
-    def multistatement(self):
+    def test_multistatement(self):
         query = 'SELECT * FROM t1; SELECT * FROM t2'
         self.assertEquals({'t1', 't2'}, self.extract_tables(query))
 
         query = 'SELECT * FROM t1; SELECT * FROM t2;'
         self.assertEquals({'t1', 't2'}, self.extract_tables(query))
+
+    def test_update_not_select(self):
+        sql = sql_parse.SupersetQuery('UPDATE t1 SET col1 = NULL')
+        self.assertEquals(False, sql.is_select())
+        self.assertEquals(False, sql.is_readonly())
+
+    def test_explain(self):
+        sql = sql_parse.SupersetQuery('EXPLAIN SELECT 1')
+
+        self.assertEquals(True, sql.is_explain())
+        self.assertEquals(False, sql.is_select())
+        self.assertEquals(True, sql.is_readonly())

tests/sqllab_tests.py

@@ -55,6 +55,12 @@ def test_sql_json(self):
         data = self.run_sql('SELECT * FROM unexistant_table', '2')
         self.assertLess(0, len(data['error']))
 
+    def test_explain(self):
+        self.login('admin')
+
+        data = self.run_sql('EXPLAIN SELECT * FROM ab_user', '1')
+        self.assertLess(0, len(data['data']))
+
     def test_sql_json_has_access(self):
         main_db = self.get_main_database(db.session)
         security_manager.add_permission_view_menu('database_access', main_db.perm)